Fix computed_attribute_setup_[both] to only update each computed attribute once

Author: dgarros

backend/infrahub/computed_attribute/tasks.py

@@ -302,15 +302,24 @@ async def computed_attribute_setup_jinja2(
 
         triggers = await gather_trigger_computed_attribute_jinja2()
 
-        for trigger in triggers:
-            if event_name != BranchDeletedEvent.event_name and trigger.branch == branch_name:
+        # 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)  # type: ignore[attr-defined]
+            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": trigger.branch,
-                        "computed_attribute_name": trigger.computed_attribute.attribute.name,
-                        "computed_attribute_kind": trigger.computed_attribute.kind,
+                        "branch_name": branch,
+                        "computed_attribute_name": attribute_name,
+                        "computed_attribute_kind": kind,
                     },
                 )
 
@@ -320,6 +329,7 @@ async def computed_attribute_setup_jinja2(
                 client=prefect_client,
                 triggers=triggers,
                 trigger_type=TriggerType.COMPUTED_ATTR_JINJA2,
+                force_update=False,
             )  # type: ignore[misc]
 
         log.info(f"{len(triggers)} Computed Attribute for Jinja2 automation configuration completed")
@@ -347,18 +357,29 @@ async def computed_attribute_setup_python(
 
         triggers_python, triggers_python_query = await gather_trigger_computed_attribute_python(db=db)
 
-        for trigger in triggers_python:
-            if event_name != BranchDeletedEvent.event_name and trigger.branch == branch_name:
-                log.info(
-                    f"Triggering update for {trigger.computed_attribute.computed_attribute.attribute.name} on {branch_name}"
-                )
+        # 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,  # type: ignore[attr-defined]
+                trigger.computed_attribute.computed_attribute.kind,  # type: ignore[attr-defined]
+                trigger.computed_attribute.computed_attribute.attribute.name,  # type: ignore[attr-defined]
+            )
+            for trigger in triggers_python
+        }
+        for branch, kind, attribute_name in unique_nodes:
+            if event_name != BranchDeletedEvent.event_name and branch == branch_name:
+                log.info(f"Triggering update for {kind}.{attribute_name} on {branch}")
                 await service.workflow.submit_workflow(
                     workflow=TRIGGER_UPDATE_PYTHON_COMPUTED_ATTRIBUTES,
                     context=context,
                     parameters={
                         "branch_name": branch_name,
-                        "computed_attribute_name": trigger.computed_attribute.computed_attribute.attribute.name,
-                        "computed_attribute_kind": trigger.computed_attribute.computed_attribute.kind,
+                        "computed_attribute_name": attribute_name,
+                        "computed_attribute_kind": kind,
                     },
                 )
 

backend/infrahub/trigger/models.py

@@ -5,8 +5,11 @@
 from typing import TYPE_CHECKING, Any
 
 from prefect.events.actions import RunDeployment
+from prefect.events.schemas.automations import (
+    Automation,  # noqa: TC002
+    Posture,
+)
 from prefect.events.schemas.automations import EventTrigger as PrefectEventTrigger
-from prefect.events.schemas.automations import Posture
 from prefect.events.schemas.events import ResourceSpecification
 from pydantic import BaseModel, Field
 
@@ -19,6 +22,13 @@
     from uuid import UUID
 
 
+class TriggerSetupReport(BaseModel):
+    created: list[TriggerDefinition] = Field(default_factory=list)
+    updated: list[TriggerDefinition] = Field(default_factory=list)
+    deleted: list[Automation] = Field(default_factory=list)
+    unchanged: list[TriggerDefinition] = Field(default_factory=list)
+
+
 class TriggerType(str, Enum):
     BUILTIN = "builtin"
     WEBHOOK = "webhook"

backend/infrahub/trigger/setup.py

@@ -5,23 +5,39 @@
 from prefect.cache_policies import NONE
 from prefect.client.orchestration import PrefectClient
 from prefect.client.schemas.filters import DeploymentFilter, DeploymentFilterName
+from prefect.events.schemas.automations import Automation
 
 from infrahub.trigger.models import TriggerDefinition
 
-from .models import TriggerType
+from .models import TriggerSetupReport, TriggerType
 
 if TYPE_CHECKING:
     from uuid import UUID
 
 
+def compare_automations(target: AutomationCore, existing: Automation) -> bool:
+    """Compare an AutomationCore with an existing Automation object to identify if they are identical or not
+
+    Return True if the target is identical to the existing automatino
+    """
+
+    target_dump = target.model_dump(exclude_defaults=True, exclude_none=True)
+    existing_dump = existing.model_dump(exclude_defaults=True, exclude_none=True, exclude={"id"})
+
+    return target_dump == existing_dump
+
+
 @task(name="trigger-setup", task_run_name="Setup triggers", cache_policy=NONE)  # type: ignore[arg-type]
 async def setup_triggers(
     client: PrefectClient,
     triggers: list[TriggerDefinition],
     trigger_type: TriggerType | None = None,
-) -> None:
+    force_update: bool = False,
+) -> TriggerSetupReport:
     log = get_run_logger()
 
+    report = TriggerSetupReport()
+
     if trigger_type:
         log.info(f"Setting up triggers of type {trigger_type.value}")
     else:
@@ -38,23 +54,24 @@ async def setup_triggers(
         )
     }
     deployments_mapping: dict[str, UUID] = {name: item.id for name, item in deployments.items()}
-    existing_automations = {item.name: item for item in await client.read_automations()}
 
     # If a trigger type is provided, narrow down the list of existing triggers to know which one to delete
+    existing_automations: dict[str, Automation] = {}
     if trigger_type:
-        trigger_automations = [
-            item.name for item in await client.read_automations() if item.name.startswith(trigger_type.value)
-        ]
+        existing_automations = {
+            item.name: item for item in await client.read_automations() if item.name.startswith(trigger_type.value)
+        }
     else:
-        trigger_automations = [item.name for item in await client.read_automations()]
+        existing_automations = {item.name: item for item in await client.read_automations()}
 
     trigger_names = [trigger.generate_name() for trigger in triggers]
+    automation_names = list(existing_automations.keys())
 
-    log.debug(f"{len(trigger_automations)} existing triggers ({trigger_automations})")
-    log.debug(f"{len(trigger_names)}  triggers to configure ({trigger_names})")
+    log.debug(f"{len(automation_names)} existing triggers ({automation_names})")
+    log.debug(f"{len(trigger_names)} triggers to configure ({trigger_names})")
 
-    to_delete = set(trigger_automations) - set(trigger_names)
-    log.debug(f"{len(trigger_names)} triggers to delete ({to_delete})")
+    to_delete = set(automation_names) - set(trigger_names)
+    log.debug(f"{len(to_delete)} triggers to delete ({to_delete})")
 
     # -------------------------------------------------------------
     # Create or Update all triggers
@@ -71,11 +88,16 @@ async def setup_triggers(
         existing_automation = existing_automations.get(trigger.generate_name(), None)
 
         if existing_automation:
-            await client.update_automation(automation_id=existing_automation.id, automation=automation)
-            log.info(f"{trigger.generate_name()} Updated")
+            if force_update or not compare_automations(target=automation, existing=existing_automation):
+                await client.update_automation(automation_id=existing_automation.id, automation=automation)
+                log.info(f"{trigger.generate_name()} Updated")
+                report.updated.append(trigger)
+            else:
+                report.unchanged.append(trigger)
         else:
             await client.create_automation(automation=automation)
             log.info(f"{trigger.generate_name()} Created")
+            report.created.append(trigger)
 
     # -------------------------------------------------------------
     # Delete Triggers that shouldn't be there
@@ -86,5 +108,8 @@ async def setup_triggers(
         if not existing_automation:
             continue
 
+        report.deleted.append(existing_automation)
         await client.delete_automation(automation_id=existing_automation.id)
         log.info(f"{item_to_delete} Deleted")
+
+    return report

backend/infrahub/trigger/tasks.py

@@ -31,7 +31,4 @@ async def trigger_configure_all(service: InfrahubServices) -> None:
         )
 
         async with get_client(sync_client=False) as prefect_client:
-            await setup_triggers(
-                client=prefect_client,
-                triggers=triggers,
-            )
+            await setup_triggers(client=prefect_client, triggers=triggers, force_update=True)

backend/infrahub/workflows/initialization.py

@@ -71,7 +71,5 @@ async def setup_task_manager() -> None:
         await setup_worker_pools(client=client)
         await setup_deployments(client=client)
         await setup_triggers(
-            client=client,
-            triggers=builtin_triggers,
-            trigger_type=TriggerType.BUILTIN,
+            client=client, triggers=builtin_triggers, trigger_type=TriggerType.BUILTIN, force_update=True
         )

backend/tests/unit/trigger/test_tasks.py

@@ -2,7 +2,7 @@
 from prefect.client.orchestration import PrefectClient, get_client
 
 from infrahub.trigger.catalogue import builtin_triggers
-from infrahub.trigger.models import TriggerType
+from infrahub.trigger.models import EventTrigger, TriggerType
 from infrahub.trigger.setup import setup_triggers
 from infrahub.workflows.initialization import setup_deployments, setup_worker_pools
 
@@ -27,12 +27,35 @@ async def init_prefect(prefect_client: PrefectClient) -> None:
 
 
 async def test_setup_triggers(prefect_client: PrefectClient, init_prefect, cleanup_automation):
-    await setup_triggers(client=prefect_client, triggers=builtin_triggers, trigger_type=TriggerType.BUILTIN)
+    report = await setup_triggers(client=prefect_client, triggers=builtin_triggers, trigger_type=TriggerType.BUILTIN)
+
+    assert len(report.deleted) == 0
+    assert len(report.updated) == 0
+    assert len(report.unchanged) == 0
+    assert len(report.created) == len(builtin_triggers)
 
     automations = await prefect_client.read_automations()
     assert len(automations) == len(builtin_triggers)
 
-    # Remove 2 Triggers and ensure that setup_triggers is working as expected
-    await setup_triggers(client=prefect_client, triggers=builtin_triggers[:-2], trigger_type=TriggerType.BUILTIN)
+    # Update 1 Trigger and remove 2 to ensure that setup_triggers is working as expected
+    builtin_triggers[0].trigger = EventTrigger(events={"new.event.name"})
+    report_after = await setup_triggers(
+        client=prefect_client, triggers=builtin_triggers[:-2], trigger_type=TriggerType.BUILTIN
+    )
+
+    assert len(report_after.deleted) == 2
+    assert len(report_after.updated) == 1
+    assert len(report_after.unchanged) == len(builtin_triggers) - 3
+    assert len(report_after.created) == 0
+
     automations = await prefect_client.read_automations()
     assert len(automations) == len(builtin_triggers[:-2])
+
+    # Ensure force_update is working properly
+    report_force = await setup_triggers(
+        client=prefect_client, triggers=builtin_triggers[:-2], trigger_type=TriggerType.BUILTIN, force_update=True
+    )
+    assert len(report_force.deleted) == 0
+    assert len(report_force.updated) == len(builtin_triggers) - 2
+    assert len(report_force.unchanged) == 0
+    assert len(report_force.created) == 0

changelog/+comp_attr_setup_dup.fixed.md

@@ -0,0 +1 @@
+Fixed issue with computed attribute that would trigger multiple updates after a schema change if the attribute reference multiple kind of nodes.