Add validator events

Author: ogenstad

backend/infrahub/core/constants/__init__.py

@@ -66,6 +66,10 @@ class EventType(InfrahubStringEnum):
     ARTIFACT_CREATED = f"{EVENT_NAMESPACE}.artifact.created"
     ARTIFACT_UPDATED = f"{EVENT_NAMESPACE}.artifact.updated"
 
+    VALIDATOR_STARTED = f"{EVENT_NAMESPACE}.validator.started"
+    VALIDATOR_PASSED = f"{EVENT_NAMESPACE}.validator.passed"
+    VALIDATOR_FAILED = f"{EVENT_NAMESPACE}.validator.failed"
+
 
 class PermissionLevel(enum.Flag):
     READ = 1

backend/infrahub/core/validators/checks_runner.py

@@ -1,14 +1,21 @@
 import asyncio
 from typing import Any, Coroutine
 
-from infrahub_sdk.node import InfrahubNode
+from infrahub_sdk.protocols import CoreValidator
 
+from infrahub.context import InfrahubContext
 from infrahub.core.constants import ValidatorConclusion, ValidatorState
 from infrahub.core.timestamp import Timestamp
+from infrahub.services import InfrahubServices
+from infrahub.validators.events import send_failed_validator, send_passed_validator
 
 
 async def run_checks_and_update_validator(
-    checks: list[Coroutine[Any, None, ValidatorConclusion]], validator: InfrahubNode
+    checks: list[Coroutine[Any, None, ValidatorConclusion]],
+    validator: CoreValidator,
+    context: InfrahubContext,
+    service: InfrahubServices,
+    proposed_change_id: str,
 ) -> None:
     """
     Execute a list of checks coroutines, and set validator fields accordingly.
@@ -22,11 +29,17 @@ async def run_checks_and_update_validator(
     validator.completed_at.value = ""
     await validator.save()
 
+    failed_early = False
+
     for earliest_task in asyncio.as_completed(checks):
         result = await earliest_task
         if validator.conclusion.value != ValidatorConclusion.FAILURE.value and result == ValidatorConclusion.FAILURE:
             validator.conclusion.value = ValidatorConclusion.FAILURE.value
+            failed_early = True
             await validator.save()
+            await send_failed_validator(
+                service=service, validator=validator, proposed_change_id=proposed_change_id, context=context
+            )
             # Continue to iterate to wait for the end of all checks
 
     validator.state.value = ValidatorState.COMPLETED.value
@@ -35,3 +48,13 @@ async def run_checks_and_update_validator(
         validator.conclusion.value = ValidatorConclusion.SUCCESS.value
 
     await validator.save()
+
+    if not failed_early:
+        if validator.conclusion.value == ValidatorConclusion.SUCCESS.value:
+            await send_passed_validator(
+                service=service, validator=validator, proposed_change_id=proposed_change_id, context=context
+            )
+        else:
+            await send_failed_validator(
+                service=service, validator=validator, proposed_change_id=proposed_change_id, context=context
+            )

backend/infrahub/events/validator_action.py

@@ -0,0 +1,61 @@
+from pydantic import Field, computed_field
+
+from infrahub.message_bus import InfrahubMessage
+
+from .constants import EVENT_NAMESPACE
+from .models import InfrahubEvent
+
+
+class ValidatorEvent(InfrahubEvent):
+    """Event generated when an validator within a pipeline has started."""
+
+    node_id: str = Field(..., description="The ID of the validator")
+    kind: str = Field(..., description="The kind of the validator")
+    proposed_change_id: str = Field(..., description="The ID of the proposed change")
+
+    def get_resource(self) -> dict[str, str]:
+        return {
+            "prefect.resource.id": self.node_id,
+            "infrahub.node.kind": self.kind,
+            "infrahub.node.id": self.node_id,
+            "infrahub.branch.name": self.meta.context.branch.name,
+        }
+
+    def get_related(self) -> list[dict[str, str]]:
+        related = super().get_related()
+        related.append(
+            {
+                "prefect.resource.id": self.proposed_change_id,
+                "prefect.resource.role": "infrahub.related.node",
+                "infrahub.node.kind": "CoreProposedChange",
+            }
+        )
+
+        return related
+
+    def get_messages(self) -> list[InfrahubMessage]:
+        return []
+
+
+class ValidatorStartedEvent(ValidatorEvent):
+    """Event generated when an validator within a pipeline has started."""
+
+    @computed_field
+    def event_name(self) -> str:
+        return f"{EVENT_NAMESPACE}.validator.started"
+
+
+class ValidatorPassedEvent(ValidatorEvent):
+    """Event generated when an validator within a pipeline has completed successfully."""
+
+    @computed_field
+    def event_name(self) -> str:
+        return f"{EVENT_NAMESPACE}.validator.passed"
+
+
+class ValidatorFailedEvent(ValidatorEvent):
+    """Event generated when an validator within a pipeline has completed successfully."""
+
+    @computed_field
+    def event_name(self) -> str:
+        return f"{EVENT_NAMESPACE}.validator.failed"

backend/infrahub/git/tasks.py

@@ -1,5 +1,12 @@
 from infrahub_sdk import InfrahubClient
-from infrahub_sdk.protocols import CoreArtifact, CoreArtifactDefinition, CoreCheckDefinition, CoreRepository
+from infrahub_sdk.protocols import (
+    CoreArtifact,
+    CoreArtifactDefinition,
+    CoreCheckDefinition,
+    CoreRepository,
+    CoreRepositoryValidator,
+    CoreUserValidator,
+)
 from infrahub_sdk.uuidt import UUIDT
 from prefect import flow, task
 from prefect.cache_policies import NONE
@@ -12,6 +19,7 @@
 from infrahub.exceptions import CheckError, RepositoryError
 from infrahub.message_bus import Meta, messages
 from infrahub.services import InfrahubServices
+from infrahub.validators.events import send_start_validator
 from infrahub.worker import WORKER_IDENTITY
 
 from ..core.manager import NodeManager
@@ -509,7 +517,9 @@ async def git_repository_diff_names_only(
     name="git-repository-user-checks-definition-trigger",
     flow_run_name="Trigger user defined checks for repository {model.repository_name}",
 )
-async def trigger_repository_user_checks_definitions(model: UserCheckDefinitionData, service: InfrahubServices) -> None:
+async def trigger_repository_user_checks_definitions(
+    model: UserCheckDefinitionData, context: InfrahubContext, service: InfrahubServices
+) -> None:
     await add_tags(branches=[model.branch_name], nodes=[model.proposed_change])
     log = get_run_logger()
 
@@ -520,7 +530,7 @@ async def trigger_repository_user_checks_definitions(model: UserCheckDefinitionD
     validator_execution_id = str(UUIDT())
     check_execution_ids: list[str] = []
     await proposed_change.validations.fetch()
-    validator = None
+    validator: CoreUserValidator | None = None
 
     for relationship in proposed_change.validations.peers:
         existing_validator = relationship.peer
@@ -541,7 +551,7 @@ async def trigger_repository_user_checks_definitions(model: UserCheckDefinitionD
         await validator.save()
     else:
         validator = await service.client.create(
-            kind=InfrahubKind.USERVALIDATOR,
+            kind=CoreUserValidator,
             data={
                 "label": f"Check: {definition.name.value}",
                 "proposed_change": model.proposed_change,
@@ -551,6 +561,10 @@ async def trigger_repository_user_checks_definitions(model: UserCheckDefinitionD
         )
         await validator.save()
 
+    await send_start_validator(
+        service=service, validator=validator, proposed_change_id=model.proposed_change, context=context
+    )
+
     if definition.targets.id:
         # Check against a group of targets
         await definition.targets.fetch()
@@ -612,14 +626,22 @@ async def trigger_repository_user_checks_definitions(model: UserCheckDefinitionD
         for model in check_models
     ]
 
-    await run_checks_and_update_validator(checks_coroutines, validator)
+    await run_checks_and_update_validator(
+        checks=checks_coroutines,
+        validator=validator,
+        context=context,
+        service=service,
+        proposed_change_id=model.proposed_change,
+    )
 
 
 @flow(
     name="git-repository-trigger-user-checks",
     flow_run_name="Evaluating user-defined checks on repository {model.repository_name}",
 )
-async def trigger_user_checks(model: TriggerRepositoryUserChecks, service: InfrahubServices) -> None:
+async def trigger_user_checks(
+    model: TriggerRepositoryUserChecks, service: InfrahubServices, context: InfrahubContext
+) -> None:
     """Request to start validation checks on a specific repository for User-defined checks."""
 
     await add_tags(branches=[model.source_branch], nodes=[model.proposed_change])
@@ -645,15 +667,19 @@ async def trigger_user_checks(model: TriggerRepositoryUserChecks, service: Infra
             branch_diff=model.branch_diff,
         )
         await service.workflow.submit_workflow(
-            workflow=GIT_REPOSITORY_USER_CHECKS_DEFINITIONS_TRIGGER, parameters={"model": user_check_definition_model}
+            workflow=GIT_REPOSITORY_USER_CHECKS_DEFINITIONS_TRIGGER,
+            context=context,
+            parameters={"model": user_check_definition_model},
         )
 
 
 @flow(
     name="git-repository-trigger-internal-checks",
     flow_run_name="Running repository checks for repository {model.repository}",
 )
-async def trigger_internal_checks(model: TriggerRepositoryInternalChecks, service: InfrahubServices) -> None:
+async def trigger_internal_checks(
+    model: TriggerRepositoryInternalChecks, service: InfrahubServices, context: InfrahubContext
+) -> None:
     """Request to start validation checks on a specific repository."""
     await add_tags(branches=[model.source_branch], nodes=[model.proposed_change])
     log = get_run_logger()
@@ -669,7 +695,7 @@ async def trigger_internal_checks(model: TriggerRepositoryInternalChecks, servic
     await repository.checks.fetch()
 
     validator_name = f"Repository Validator: {repository.name.value}"
-    validator = None
+    validator: CoreRepositoryValidator | None = None
     for relationship in proposed_change.validations.peers:
         existing_validator = relationship.peer
 
@@ -687,7 +713,7 @@ async def trigger_internal_checks(model: TriggerRepositoryInternalChecks, servic
         await validator.save()
     else:
         validator = await service.client.create(
-            kind=InfrahubKind.REPOSITORYVALIDATOR,
+            kind=CoreRepositoryValidator,
             data={
                 "label": validator_name,
                 "proposed_change": model.proposed_change,
@@ -696,6 +722,10 @@ async def trigger_internal_checks(model: TriggerRepositoryInternalChecks, servic
         )
         await validator.save()
 
+    await send_start_validator(
+        service=service, validator=validator, proposed_change_id=model.proposed_change, context=context
+    )
+
     check_execution_id = str(UUIDT())
     check_execution_ids.append(check_execution_id)
     log.info("Adding check for merge conflict")
@@ -718,7 +748,13 @@ async def trigger_internal_checks(model: TriggerRepositoryInternalChecks, servic
         expected_return=ValidatorConclusion,
     )
 
-    await run_checks_and_update_validator(checks=[check_coroutine], validator=validator)
+    await run_checks_and_update_validator(
+        checks=[check_coroutine],
+        validator=validator,
+        context=context,
+        service=service,
+        proposed_change_id=model.proposed_change,
+    )
 
 
 @flow(

backend/infrahub/message_bus/messages/finalize_validator_execution.py

@@ -1,5 +1,6 @@
 from pydantic import Field
 
+from infrahub.context import InfrahubContext
 from infrahub.message_bus import InfrahubMessage
 
 
@@ -10,3 +11,5 @@ class FinalizeValidatorExecution(InfrahubMessage):
     validator_execution_id: str = Field(..., description="The id of current execution of the associated validator")
     start_time: str = Field(..., description="Start time when the message was first created")
     validator_type: str = Field(..., description="The type of validator to complete")
+    context: InfrahubContext = Field(..., description="The Infrahub context")
+    proposed_change: str = Field(..., description="The ID of the proposed change")

backend/infrahub/message_bus/messages/request_generatordefinition_check.py

@@ -1,5 +1,6 @@
 from pydantic import ConfigDict, Field
 
+from infrahub.context import InfrahubContext
 from infrahub.generators.models import ProposedChangeGeneratorDefinition
 from infrahub.message_bus import InfrahubMessage
 from infrahub.message_bus.types import ProposedChangeBranchDiff
@@ -16,3 +17,4 @@ class RequestGeneratorDefinitionCheck(InfrahubMessage):
     source_branch: str = Field(..., description="The source branch")
     source_branch_sync_with_git: bool = Field(..., description="Indicates if the source branch should sync with git")
     destination_branch: str = Field(..., description="The target branch")
+    context: InfrahubContext = Field(..., description="The Infrahub context")

backend/infrahub/message_bus/operations/finalize/validator.py

@@ -1,11 +1,22 @@
+from infrahub_sdk.protocols import (
+    CoreArtifactValidator,
+    CoreDataValidator,
+    CoreGeneratorValidator,
+    CoreRepositoryValidator,
+    CoreSchemaValidator,
+    CoreUserValidator,
+    CoreValidator,
+)
 from prefect import flow
 
 from infrahub import config
+from infrahub.core.constants import InfrahubKind, ValidatorConclusion
 from infrahub.core.timestamp import Timestamp
 from infrahub.log import get_logger
 from infrahub.message_bus import messages
 from infrahub.message_bus.types import KVTTL, MessageTTL
 from infrahub.services import InfrahubServices
+from infrahub.validators.events import send_failed_validator, send_passed_validator
 
 log = get_logger()
 
@@ -20,7 +31,8 @@ async def execution(message: messages.FinalizeValidatorExecution, service: Infra
 
     The message will get rescheduled until the timeout has exceeded or until all checks are accounted for.
     """
-    validator = await service.client.get(kind=message.validator_type, id=message.validator_id)
+    validator_type = get_validator_type(validator_type=message.validator_type)
+    validator = await service.client.get(kind=validator_type, id=message.validator_id)
     checks_key = f"validator_execution_id:{message.validator_execution_id}:checks"
     current_conclusion = validator.conclusion.value
     if validator.state.value != "in_progress":
@@ -81,3 +93,41 @@ async def execution(message: messages.FinalizeValidatorExecution, service: Infra
     validator.completed_at.value = Timestamp().to_string()
     validator.conclusion.value = conclusion
     await validator.save()
+    if validator.conclusion.value == ValidatorConclusion.SUCCESS.value:
+        await send_passed_validator(
+            service=service, validator=validator, proposed_change_id=message.proposed_change, context=message.context
+        )
+    else:
+        await send_failed_validator(
+            service=service, validator=validator, proposed_change_id=message.proposed_change, context=message.context
+        )
+
+
+def get_validator_type(
+    validator_type: str,
+) -> (
+    type[CoreArtifactValidator]
+    | type[CoreDataValidator]
+    | type[CoreGeneratorValidator]
+    | type[CoreRepositoryValidator]
+    | type[CoreSchemaValidator]
+    | type[CoreUserValidator]
+    | type[CoreValidator]
+):
+    match validator_type:
+        case InfrahubKind.USERVALIDATOR:
+            validator_kind = CoreUserValidator
+        case InfrahubKind.SCHEMAVALIDATOR:
+            validator_kind = CoreSchemaValidator
+        case InfrahubKind.GENERATORVALIDATOR:
+            validator_kind = CoreGeneratorValidator
+        case InfrahubKind.REPOSITORYVALIDATOR:
+            validator_kind = CoreRepositoryValidator
+        case InfrahubKind.DATAVALIDATOR:
+            validator_kind = CoreDataValidator
+        case InfrahubKind.ARTIFACTVALIDATOR:
+            validator_kind = CoreArtifactValidator
+        case _:
+            validator_kind = CoreValidator
+
+    return validator_kind