add FreshnessStateChange event

Author: anuthebananu

python_modules/dagster/dagster/_core/definitions/freshness.py

@@ -28,6 +28,17 @@ class FreshnessStateEvaluation:
     freshness_state: FreshnessState
 
 
+@whitelist_for_serdes
+@record
+class FreshnessStateChange:
+    """Event that is emitted when the freshness state of an asset changes."""
+
+    key: AssetKey
+    new_state: FreshnessState
+    previous_state: FreshnessState
+    state_change_timestamp: float
+
+
 INTERNAL_FRESHNESS_POLICY_METADATA_KEY = "dagster/internal_freshness_policy"
 
 

python_modules/dagster/dagster/_core/events/__init__.py

@@ -67,7 +67,7 @@
 
 if TYPE_CHECKING:
     from dagster._core.definitions.events import ObjectStoreOperation
-    from dagster._core.definitions.freshness import FreshnessStateEvaluation
+    from dagster._core.definitions.freshness import FreshnessStateChange, FreshnessStateEvaluation
     from dagster._core.execution.plan.plan import ExecutionPlan
     from dagster._core.execution.plan.step import StepKind
 
@@ -95,6 +95,7 @@
     "AssetFailedToMaterializeData",
     "RunEnqueuedData",
     "FreshnessStateEvaluation",
+    "FreshnessStateChange",
 ]
 
 
@@ -172,6 +173,7 @@ class DagsterEventType(str, Enum):
     LOGS_CAPTURED = "LOGS_CAPTURED"
 
     FRESHNESS_STATE_EVALUATION = "FRESHNESS_STATE_EVALUATION"
+    FRESHNESS_STATE_CHANGE = "FRESHNESS_STATE_CHANGE"
 
 
 EVENT_TYPE_TO_DISPLAY_STRING = {

python_modules/dagster/dagster/_core/instance/__init__.py

@@ -38,7 +38,11 @@
 )
 from dagster._core.definitions.data_version import extract_data_provenance_from_entry
 from dagster._core.definitions.events import AssetKey, AssetObservation
-from dagster._core.definitions.freshness import FreshnessStateEvaluation, FreshnessStateRecord
+from dagster._core.definitions.freshness import (
+    FreshnessStateChange,
+    FreshnessStateEvaluation,
+    FreshnessStateRecord,
+)
 from dagster._core.definitions.partition_key_range import PartitionKeyRange
 from dagster._core.errors import (
     DagsterHomeNotSetError,
@@ -3479,6 +3483,7 @@ def report_runless_asset_event(
             "AssetObservation",
             "AssetCheckEvaluation",
             "FreshnessStateEvaluation",
+            "FreshnessStateChange",
         ],
     ):
         """Record an event log entry related to assets that does not belong to a Dagster run."""
@@ -3502,6 +3507,9 @@ def report_runless_asset_event(
         elif isinstance(asset_event, FreshnessStateEvaluation):
             event_type_value = DagsterEventType.FRESHNESS_STATE_EVALUATION.value
             data_payload = asset_event
+        elif isinstance(asset_event, FreshnessStateChange):
+            event_type_value = DagsterEventType.FRESHNESS_STATE_CHANGE.value
+            data_payload = asset_event
         else:
             raise DagsterInvariantViolationError(
                 f"Received unexpected asset event type {asset_event}, expected"