Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Ele 1706 add to freshness result criteria #1175

Merged
merged 10 commits into from
Sep 28, 2023
Merged

Ele 1706 add to freshness result criteria #1175

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
00bbd8c
criteria + normalized status
NoyaArie Sep 21, 2023
0cf5854
use error
NoyaArie Sep 26, 2023
eed4ae7
cleanup
NoyaArie Sep 26, 2023
71cd389
fix mock objects
NoyaArie Sep 26, 2023
b472601
query new field
NoyaArie Sep 27, 2023
5af234f
add description to alert
NoyaArie Sep 27, 2023
bc0b99c
Revert "add description to alert"
NoyaArie Sep 27, 2023
5650bcc
Revert "query new field"
NoyaArie Sep 27, 2023
3a7483d
Merge branch 'master' into ele-1706-add-result-description-to-source-…
NoyaArie Sep 27, 2023
ad776b1
fix tests
NoyaArie Sep 28, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 18 additions & 17 deletions elementary/monitor/alerts/source_freshness.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,10 @@ def __init__(
max_loaded_at_time_ago_in_s: Optional[float],
source_name: str,
identifier: str,
freshness_error_after: str,
freshness_warn_after: str,
freshness_filter: str,
normalized_status: str,
error_after: str,
warn_after: str,
filter: Optional[str],
path: str,
error: str,
**kwargs,
Expand All @@ -42,9 +43,10 @@ def __init__(
self.max_loaded_at_time_ago_in_s = max_loaded_at_time_ago_in_s
self.source_name = source_name
self.identifier = identifier
self.freshness_error_after = freshness_error_after
self.freshness_warn_after = freshness_warn_after
self.freshness_filter = freshness_filter
self.normalized_status = normalized_status
self.error_after = error_after
self.warn_after = warn_after
self.filter = filter
self.path = path
self.error = error
self.alerts_table = SourceFreshnessAlert.TABLE_NAME
Expand All @@ -55,7 +57,7 @@ def to_slack(self, is_slack_workflow: bool = False) -> SlackMessageSchema:
subscribers = self.slack_message_builder.prettify_and_dedup_list(
self.subscribers or []
)
icon = self.slack_message_builder.get_slack_status_icon(self.status)
icon = self.slack_message_builder.get_slack_status_icon(self.normalized_status)

title = [
self.slack_message_builder.create_header_block(
Expand All @@ -68,7 +70,7 @@ def to_slack(self, is_slack_workflow: bool = False) -> SlackMessageSchema:
self.slack_message_builder.create_context_block(
[
f"*Source:* {self.source_name}.{self.identifier} |",
f"*Status:* {self.status}",
f"*Status:* {self.normalized_status}",
],
),
self.slack_message_builder.create_context_block(
Expand All @@ -84,7 +86,7 @@ def to_slack(self, is_slack_workflow: bool = False) -> SlackMessageSchema:
self.slack_message_builder.create_context_block(
[
f"*Source:* {self.source_name}.{self.identifier} |",
f"*Status:* {self.status} |",
f"*Status:* {self.normalized_status} |",
f"*{self.detected_at_str}*",
],
),
Expand Down Expand Up @@ -123,32 +125,31 @@ def to_slack(self, is_slack_workflow: bool = False) -> SlackMessageSchema:
)

configuration = []
if self.freshness_error_after:

if self.error_after:
configuration.append(
self.slack_message_builder.create_context_block(["*Error after*"])
)
configuration.append(
self.slack_message_builder.create_text_section_block(
f"`{self.freshness_error_after}`"
f"`{self.error_after}`"
)
)
if self.freshness_warn_after:
if self.warn_after:
configuration.append(
self.slack_message_builder.create_context_block(["*Warn after*"])
)
configuration.append(
self.slack_message_builder.create_text_section_block(
f"`{self.freshness_warn_after}`"
f"`{self.warn_after}`"
)
)
if self.freshness_filter:
if self.filter:
configuration.append(
self.slack_message_builder.create_context_block(["*Filter*"])
)
configuration.append(
self.slack_message_builder.create_text_section_block(
f"`{self.freshness_filter}`"
)
self.slack_message_builder.create_text_section_block(f"`{self.filter}`")
)
if self.path:
configuration.append(
Expand Down
37 changes: 18 additions & 19 deletions elementary/monitor/dbt_project/macros/alerts/get_source_freshness_alerts.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,24 +2,14 @@
-- depends_on: {{ ref('alerts_source_freshness') }}
{% set select_pending_alerts_query %}
with alerts_in_time_limit as (
select * from {{ ref('alerts_source_freshness') }}
select *, {{ elementary_cli.normalized_source_freshness_status()}} from {{ ref('alerts_source_freshness') }}
where {{ elementary.edr_cast_as_timestamp('detected_at') }} >= {{ elementary_cli.get_alerts_time_limit(days_back) }}
),

models as (
select * from {{ ref('elementary', 'dbt_models') }}
),

sources as (
select * from {{ ref('elementary', 'dbt_sources') }}
),

artifacts_meta as (
select unique_id, meta from models
union all
select unique_id, meta from sources
),

extended_alerts as (
select
alerts_in_time_limit.alert_id,
Expand All @@ -28,6 +18,7 @@
alerts_in_time_limit.detected_at,
alerts_in_time_limit.max_loaded_at_time_ago_in_s,
alerts_in_time_limit.status,
alerts_in_time_limit.normalized_status,
alerts_in_time_limit.error,
alerts_in_time_limit.unique_id,
{# Currently alert_class_id equals to unique_id - might change in the future so we return both #}
Expand All @@ -36,9 +27,9 @@
alerts_in_time_limit.schema_name,
alerts_in_time_limit.source_name,
alerts_in_time_limit.identifier,
alerts_in_time_limit.freshness_error_after,
alerts_in_time_limit.freshness_warn_after,
alerts_in_time_limit.freshness_filter,
alerts_in_time_limit.warn_after,
alerts_in_time_limit.error_after,
alerts_in_time_limit.filter,
alerts_in_time_limit.tags,
alerts_in_time_limit.meta,
alerts_in_time_limit.owner,
Expand All @@ -51,9 +42,12 @@
else suppression_status
end as suppression_status,
alerts_in_time_limit.sent_at,
artifacts_meta.meta as model_meta
sources.meta as model_meta,
sources.freshness_error_after,
sources.freshness_warn_after,
sources.freshness_filter
from alerts_in_time_limit
left join artifacts_meta on alerts_in_time_limit.unique_id = artifacts_meta.unique_id
left join sources on alerts_in_time_limit.unique_id = sources.unique_id
)

select *
Expand All @@ -65,6 +59,10 @@
{% set alerts_dicts = elementary.agate_to_dicts(alerts_agate) %}
{% set pending_alerts = [] %}
{% for alert_dict in alerts_dicts %}
{% set error_after = alert_dict.get('error_after') %}
{% set warn_after = alert_dict.get('warn_after') %}
{% set filter = alert_dict.get('filter') %}

{% set pending_alert_dict = {'id': alert_dict.get('alert_id'),
'model_unique_id': alert_dict.get('unique_id'),
'alert_class_id': alert_dict.get('alert_class_id'),
Expand All @@ -76,10 +74,11 @@
'schema_name': alert_dict.get('schema_name'),
'source_name': alert_dict.get('source_name'),
'identifier': alert_dict.get('identifier'),
'freshness_error_after': alert_dict.get('freshness_error_after'),
'freshness_warn_after': alert_dict.get('freshness_warn_after'),
'freshness_filter': alert_dict.get('freshness_filter'),
'error_after': error_after if error_after is not none else alert_dict.get('freshness_error_after'),
'warn_after': warn_after if warn_after is not none else alert_dict.get('freshness_warn_after'),
'filter': filter if filter is not none else alert_dict.get('freshness_filter'),
'status': alert_dict.get('status'),
'normalized_status': alert_dict.get('normalized_status'),
'owners': alert_dict.get('owner'),
'path': alert_dict.get('path'),
'error': alert_dict.get('error'),
Expand Down
7 changes: 7 additions & 0 deletions elementary/monitor/dbt_project/macros/utils/normalized_source_freshness_status.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
{% macro normalized_source_freshness_status() %}
case
when status = 'error' then 'fail'
when status = 'runtime error' then 'error'
else status
end as normalized_status
{% endmacro %}
21 changes: 12 additions & 9 deletions tests/unit/monitor/api/alerts/test_alert_filters.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,14 +171,15 @@ def initial_alerts():
tags='["one", "two"]',
owners='["jeff", "john"]',
status="error",
normalized_status="fail",
snapshotted_at="2023-08-15T12:26:06.884065+00:00",
max_loaded_at="1969-12-31T00:00:00+00:00",
max_loaded_at_time_ago_in_s=1692188766.884065,
source_name="elementary_integration_tests",
identifier="any_type_column_anomalies_validation",
freshness_error_after='{"count": null, "period": null}',
freshness_warn_after='{"count": 1, "period": "minute"}',
freshness_filter="null",
error_after='{"count": null, "period": null}',
warn_after='{"count": 1, "period": "minute"}',
filter="null",
error="problemz",
),
SourceFreshnessAlert(
Expand All @@ -196,14 +197,15 @@ def initial_alerts():
tags='["one", "two"]',
owners='["jeff", "john"]',
status="warn",
normalized_status="warn",
snapshotted_at="2023-08-15T12:26:06.884065+00:00",
max_loaded_at="1969-12-31T00:00:00+00:00",
max_loaded_at_time_ago_in_s=1692188766.884065,
source_name="elementary_integration_tests",
identifier="any_type_column_anomalies_validation",
freshness_error_after='{"count": null, "period": null}',
freshness_warn_after='{"count": 1, "period": "minute"}',
freshness_filter="null",
error_after='{"count": null, "period": null}',
warn_after='{"count": 1, "period": "minute"}',
filter="null",
error="problemz",
),
SourceFreshnessAlert(
Expand All @@ -221,14 +223,15 @@ def initial_alerts():
tags='["one", "two"]',
owners='["jeff", "john"]',
status="runtime error",
normalized_status="error",
snapshotted_at="2023-08-15T12:26:06.884065+00:00",
max_loaded_at="1969-12-31T00:00:00+00:00",
max_loaded_at_time_ago_in_s=1692188766.884065,
source_name="elementary_integration_tests",
identifier="any_type_column_anomalies_validation",
freshness_error_after='{"count": null, "period": null}',
freshness_warn_after='{"count": 1, "period": "minute"}',
freshness_filter="null",
error_after='{"count": null, "period": null}',
warn_after='{"count": 1, "period": "minute"}',
filter="null",
error="problemz",
),
]
Expand Down
Loading