flip/cosmos-models
1
0
Fork 0
mirror of https://github.com/FlipsideCrypto/cosmos-models.git synced 2026-02-06 14:26:54 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

observe

Browse Source
This commit is contained in:
Eric Laurello 2023-07-11 08:53:33 -04:00
parent 9af8f0a6e6
commit b96b335242
8 changed files with 290 additions and 179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
.github/workflows/dbt_run_observability_monthly.yml vendored Normal file
View File

@ -0,0 +1,34 @@
name: dbt_run_observability_monthly
run-name: dbt_run_observability_monthly
on:
workflow_dispatch:
schedule:
- cron: '20 0 1 * *'
env:
USE_VARS: "${{ vars.USE_VARS }}"
DBT_PROFILES_DIR: "${{ vars.DBT_PROFILES_DIR }}"
DBT_VERSION: "${{ vars.DBT_VERSION }}"
ACCOUNT: "${{ vars.ACCOUNT }}"
ROLE: "${{ vars.ROLE }}"
USER: "${{ vars.USER }}"
PASSWORD: "${{ secrets.PASSWORD }}"
REGION: "${{ vars.REGION }}"
DATABASE: "${{ vars.DATABASE }}"
WAREHOUSE: "${{ vars.WAREHOUSE }}"
SCHEMA: "${{ vars.SCHEMA }}"
concurrency:
group: ${{ github.workflow }}
jobs:
called_workflow_template:
uses: FlipsideCrypto/analytics-workflow-templates/.github/workflows/dbt_run_template.yml@main
with:
dbt_command: |
dbt run -m models/silver/_observability/silver_observability* --vars "OBSERV_FULL_TEST: true"
environment: workflow_prod
warehouse: ${{ vars.WAREHOUSE }}
secrets: inherit

1
dbt_project.yml
View File

@ -57,3 +57,4 @@ vars:
STREAMLINE_INVOKE_STREAMS: False
STREAMLINE_USE_DEV_FOR_EXTERNAL_TABLES: False
"UPDATE_SNOWFLAKE_TAGS": True
OBSERV_FULL_TEST: FALSE

132
models/silver/_observability/silver_observability__block_completeness.sql Normal file
View File

@ -0,0 +1,132 @@
{{ config(
materialized = 'incremental',
full_refresh = false
) }}
WITH source AS (
SELECT
block_id,
block_timestamp,
LAG(
block_id,
1
) over (
ORDER BY
block_id ASC
) AS prev_BLOCK_ID
FROM
{{ ref('silver__blocks') }} A
WHERE
block_timestamp < DATEADD(
HOUR,
-24,
SYSDATE()
)
{% if is_incremental() %}
AND (
block_timestamp >= DATEADD(
HOUR,
-96,(
SELECT
MAX(
max_block_timestamp
)
FROM
{{ this }}
)
)
OR ({% if var('OBSERV_FULL_TEST') %}
block_id >= 0
{% else %}
block_id >= (
SELECT
MIN(VALUE) - 1
FROM
(
SELECT
blocks_impacted_array
FROM
{{ this }}
qualify ROW_NUMBER() over (
ORDER BY
test_timestamp DESC) = 1), LATERAL FLATTEN(input => blocks_impacted_array))
{% endif %})
)
{% endif %}
),
block_gen AS (
SELECT
_id AS block_id
FROM
{{ source(
'crosschain_silver',
'number_sequence'
) }}
WHERE
_id BETWEEN (
SELECT
MIN(block_id)
FROM
source
)
AND (
SELECT
MAX(block_number)
FROM
source
)
)
SELECT
'blocks' AS test_name,
MIN(
b.block_id
) AS min_block,
MAX(
b.block_id
) AS max_block,
MIN(
b.block_timestamp
) AS min_block_timestamp,
MAX(
b.block_timestamp
) AS max_block_timestamp,
COUNT(1) AS blocks_tested,
COUNT(
CASE
WHEN C.block_id IS NOT NULL THEN A.block_id
END
) AS blocks_impacted_count,
ARRAY_AGG(
CASE
WHEN C.block_id IS NOT NULL THEN A.block_id
END
) within GROUP (
ORDER BY
A.block_id
) AS blocks_impacted_array,
ARRAY_AGG(
DISTINCT CASE
WHEN C.block_id IS NOT NULL THEN OBJECT_CONSTRUCT(
'prev_block_id',
C.prev_block_id,
'block_id',
C.block_id
)
END
) AS test_failure_details,
SYSDATE() AS test_timestamp
FROM
block_gen A
LEFT JOIN source b
ON A.block_id = b.block_id
LEFT JOIN source C
ON A.block_id > C.prev_BLOCK_ID
AND A.block_id < C.block_id
AND C.block_id - C.prev_BLOCK_ID <> 1
WHERE
COALESCE(
b.block_id,
C.block_id
) IS NOT NULL

25
models/silver/_observability/silver_observability__block_gaps.yml → models/silver/_observability/silver_observability__block_completeness.yml
View File

@ -1,11 +1,11 @@
version: 2
models:
- name: silver_observability__block_gaps
description: Records of all blocks block gaps with a timestamp the test was run
- name: silver_observability__block_completeness
description: Records of all blocks block gaps (missing blocks) with a timestamp the test was run
tests:
- dbt_utils.unique_combination_of_columns:
combination_of_columns:
- _INSERTED_TIMESTAMP
- TEST_TIMESTAMP
columns:
- name: MIN_BLOCK
description: The lowest block id in the test
@ -38,29 +38,36 @@ models:
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- TIMESTAMP_NTZ
- name: TOTAL_BLOCKS
- name: BLOCKS_TESTED
description: Count of blocks in the test
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- NUMBER
- name: BLOCK_GAPS
- name: BLOCKS_IMPACTED_COUNT
description: Count of block gaps in the test
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- NUMBER
- name: BLOCK_GAPS_DETAILS
description: gap blocks
- name: BLOCKS_IMPACTED_ARRAY
description: Array of affected blocks
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- ARRAY
- name: _INSERTED_TIMESTAMP
description: "{{ doc('inserted_timestamp') }}"
- name: TEST_FAILURE_DETAILS
description: Array of details of the failure
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- ARRAY
- name: TEST_TIMESTAMP
description: When the test was run
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:

70
models/silver/_observability/silver_observability__block_gaps.sql
View File

@ -1,70 +0,0 @@
{{ config(
materialized = 'incremental',
full_refresh = false
) }}
WITH source AS (
SELECT
block_id,
block_timestamp,
LAG(
block_id,
1
) over (
ORDER BY
block_id ASC
) AS prev_BLOCK_ID
FROM
{{ ref('silver__blocks') }}
{% if is_incremental() %}
WHERE
(
block_timestamp :: DATE >= (
SELECT
MAX(
max_block_timestamp
) :: DATE -3
FROM
{{ this }}
)
OR (
(
SELECT
block_gaps
FROM
{{ this }}
qualify(ROW_NUMBER() over(
ORDER BY
max_block_timestamp DESC) = 1)
) <> 0
)
)
{% endif %}
)
SELECT
MIN(block_id) AS min_block,
MAX(block_id) AS max_block,
MIN(block_timestamp) AS min_block_timestamp,
MAX(block_timestamp) AS max_block_timestamp,
COUNT(1) AS total_blocks,
SUM(
CASE
WHEN block_id - prev_BLOCK_ID <> 1 THEN 1
ELSE 0
END
) AS block_gaps,
ARRAY_AGG(
CASE
WHEN block_id - prev_BLOCK_ID <> 1 THEN OBJECT_CONSTRUCT(
'prev_block_id',
prev_block_id,
'block_id',
block_id
)
END
) AS block_gaps_details,
SYSDATE() AS _inserted_timestamp
FROM
source

181
models/silver/_observability/silver_observability__transaction_completeness.sql
View File

@ -3,12 +3,51 @@
full_refresh = false
) }}
WITH max_silver AS (
WITH rel_blocks AS (
SELECT
MAX(block_id) AS max_block_id
block_id,
block_timestamp
FROM
{{ ref('silver__transactions') }}
{{ ref('silver__blocks') }}
WHERE
block_timestamp < DATEADD(
HOUR,
-24,
SYSDATE()
)
{% if is_incremental() %}
AND (
block_timestamp >= DATEADD(
HOUR,
-96,(
SELECT
MAX(
max_block_timestamp
)
FROM
{{ this }}
)
)
OR ({% if var('OBSERV_FULL_TEST') %}
block_id >= 0
{% else %}
block_id >= (
SELECT
MIN(VALUE) - 1
FROM
(
SELECT
blocks_impacted_array
FROM
{{ this }}
qualify ROW_NUMBER() over (
ORDER BY
test_timestamp DESC) = 1), LATERAL FLATTEN(input => blocks_impacted_array))
{% endif %})
)
{% endif %}
),
bronze AS (
SELECT
@ -17,52 +56,36 @@ bronze AS (
t.value :hash :: STRING AS tx_id
FROM
{{ ref('bronze__tx_search') }} A
JOIN {{ ref('silver__blocks') }}
b
JOIN rel_blocks b
ON A.block_id = b.block_id
JOIN TABLE(FLATTEN(DATA :result :txs)) t
WHERE
A.block_id <= (
SELECT
max_block_id
FROM
max_silver
)
AND b.block_id <= (
SELECT
max_block_id
FROM
max_silver
)
{% if is_incremental() %}
AND (
A._inserted_timestamp >= CURRENT_DATE - 7
AND b.block_timestamp :: DATE >= (
SELECT
MAX(
max_block_timestamp
) :: DATE -3
FROM
{{ this }}
)
WHERE
A._inserted_timestamp >= CURRENT_DATE - 14
OR (
(
SELECT
blocks_missing_transactions
FROM
{{ this }}
qualify(ROW_NUMBER() over(
ORDER BY
max_block_timestamp DESC) = 1)
) <> 0
)
)
SELECT
MIN(VALUE) - 1
FROM
(
SELECT
blocks_impacted_array
FROM
{{ this }}
qualify ROW_NUMBER() over (
ORDER BY
test_timestamp DESC
) = 1
),
LATERAL FLATTEN(
input => blocks_impacted_array
)
) IS NOT NULL
{% endif %}
qualify(ROW_NUMBER() over(PARTITION BY tx_id
qualify(ROW_NUMBER() over(PARTITION BY A.block_id, tx_id
ORDER BY
b._inserted_timestamp DESC) = 1)
A._inserted_timestamp DESC) = 1)
),
bronze_count AS (
SELECT
@ -85,38 +108,21 @@ bronze_api AS (
FROM
{{ ref('silver__blockchain') }}
WHERE
block_id <= (
block_timestamp BETWEEN (
SELECT
max_block_id
MIN(block_timestamp)
FROM
max_silver
rel_blocks
)
{% if is_incremental() %}
AND (
block_timestamp :: DATE >= (
SELECT
MAX(
max_block_timestamp
) :: DATE -3
FROM
{{ this }}
)
OR (
(
AND (
SELECT
blocks_missing_transactions
MAX(block_timestamp)
FROM
{{ this }}
qualify(ROW_NUMBER() over(
ORDER BY
max_block_timestamp DESC) = 1)
) <> 0
)
)
{% endif %}
rel_blocks
)
)
SELECT
'transactions' AS test_name,
MIN(
A.block_id
) AS min_block,
@ -129,7 +135,7 @@ SELECT
MAX(
A.block_timestamp
) AS max_block_timestamp,
COUNT(1) AS total_blocks,
COUNT(1) AS blocks_tested,
SUM(
CASE
WHEN COALESCE(
@ -138,7 +144,18 @@ SELECT
) - A.num_txs <> 0 THEN 1
ELSE 0
END
) AS blocks_missing_transactions,
) AS blocks_impacted_count,
ARRAY_AGG(
CASE
WHEN COALESCE(
b.num_txs,
0
) - A.num_txs <> 0 THEN A.block_id
END
) within GROUP (
ORDER BY
A.block_id
) AS blocks_impacted_array,
SUM(
ABS(
COALESCE(
@ -146,7 +163,7 @@ SELECT
0
) - A.num_txs
)
) AS total_missing_transactions,
) AS transactions_impacted_count,
ARRAY_AGG(
CASE
WHEN COALESCE(
@ -174,30 +191,8 @@ SELECT
) within GROUP(
ORDER BY
A.block_id
) AS missing_transactions_details,
SYSDATE() AS _inserted_timestamp {# SELECT
A.block_id AS min_block,
A.block_id AS max_block,
A.block_timestamp AS min_block_timestamp,
A.block_timestamp AS max_block_timestamp,
1 AS total_blocks,
CASE
WHEN A.num_txs - b.num_txs <> 0 THEN 1
ELSE 0
END AS blocks_missing_transactions,
CASE
WHEN A.num_txs - b.num_txs <> 0 THEN OBJECT_CONSTRUCT(
'block_id',
A.block_id,
'diff',
A.num_txs - b.num_txs,
'blockchain_num_txs',
b.num_txs,
'bronze_num_txs',
A.num_txs
)
END AS missing_transactions_details,
SYSDATE() AS _inserted_timestamp #}
) AS test_failure_details,
SYSDATE() AS test_timestamp
FROM
bronze_api A
LEFT JOIN bronze_count b

21
models/silver/_observability/silver_observability__transaction_completeness.yml
View File

@ -5,7 +5,7 @@ models:
tests:
- dbt_utils.unique_combination_of_columns:
combination_of_columns:
- _INSERTED_TIMESTAMP
- TEST_TIMESTAMP
columns:
- name: MIN_BLOCK
description: The lowest block id in the test
@ -38,36 +38,43 @@ models:
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- TIMESTAMP_NTZ
- name: TOTAL_BLOCKS
- name: BLOCKS_TESTED
description: Count of blocks in the test
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- NUMBER
- name: BLOCKS_MISSING_TRANSACTIONS
- name: BLOCKS_IMPACTED_COUNT
description: Count of blocks with missing transactions in the test
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- NUMBER
- name: TOTAL_MISSING_TRANSACTIONS
- name: BLOCKS_IMPACTED_ARRAY
description: Array of affected blocks
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- ARRAY
- name: TRANSACTIONS_IMPACTED_COUNT
description: Total count of missing transactions in the test
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- NUMBER
- name: MISSING_TRANSACTIONS_DETAILS
- name: TEST_FAILURE_DETAILS
description: blocks with missing transactions with the number of missing transactions
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:
column_type_list:
- ARRAY
- name: _INSERTED_TIMESTAMP
description: "{{ doc('inserted_timestamp') }}"
- name: TEST_TIMESTAMP
description: When the test was run
tests:
- not_null
- dbt_expectations.expect_column_values_to_be_in_type_list:

5
models/sources.yml
View File

@ -19,3 +19,8 @@ sources:
schema: bronze_api
tables:
- name: blockchain
- name: crosschain_silver
database: "{{ 'crosschain' if target.database == 'COSMOS' else 'crosschain_dev' }}"
schema: silver
tables:
- name: number_sequence