Ele 1453 implement metrics collection

integration_tests/integration_tests/dbt_project/models/metrics_incremental.sql

@@ -0,0 +1,3 @@
+{{ config(materialized="incremental") }}
+
+select * from {{ source("test_data", "metrics_seed2") }}

integration_tests/integration_tests/dbt_project/models/metrics_table.sql

@@ -0,0 +1,3 @@
+{{ config(materialized="table") }}
+
+select * from {{ source("test_data", "metrics_seed1") }}

integration_tests/integration_tests/dbt_project/models/metrics_view.sql

@@ -0,0 +1,5 @@
+{{ config(materialized="view") }}
+
+select * from {{ source("test_data", "metrics_seed1") }}
+union all
+select * from {{ source("test_data", "metrics_seed2") }}

integration_tests/integration_tests/dbt_project/models/test_data.yaml

@@ -4,4 +4,5 @@ sources:
   - name: test_data
     schema: "{{ target.schema }}"
     tables:
-      - name: dummy
+      - name: metrics_seed1
+      - name: metrics_seed2

integration_tests/integration_tests/dbt_project/selectors.yml

@@ -4,3 +4,6 @@ selectors:
 
   - name: one
     definition: one
+
+  - name: metrics
+    definition: metrics_*

integration_tests/integration_tests/tests/test_metrics.py

@@ -0,0 +1,38 @@
+import random
+from datetime import datetime
+
+from data_generator import DATE_FORMAT, generate_dates
+from dbt_project import DbtProject
+
+
+def test_metrics(dbt_project: DbtProject):
+    data1 = [
+        {"updated_at": date.strftime(DATE_FORMAT)}
+        for date in generate_dates(base_date=datetime.now())
+        for _ in range(random.randint(-5, 20))
+    ]
+    data2 = [
+        {"created_at": date.strftime(DATE_FORMAT)}
+        for date in generate_dates(base_date=datetime.now())
+        for _ in range(random.randint(0, 20))
+    ]
+    dbt_project.seed(data1, "metrics_seed1")
+    dbt_project.seed(data2, "metrics_seed2")
+    dbt_project.dbt_runner.run(selector="metrics")
+
+    remaining_models_to_row_count = {
+        "metrics_view": len(data1) + len(data2),
+        "metrics_table": len(data1),
+        "metrics_incremental": len(data2),
+    }
+    for metric in dbt_project.read_table("data_monitoring_metrics"):
+        for model_name, row_count in remaining_models_to_row_count.items():
+            if (
+                model_name in metric["full_table_name"]
+                and metric["metric_name"] == "row_count"
+            ):
+                assert metric["metric_value"] == row_count
+                remaining_models_to_row_count.pop(model_name)
+                break
+
+    assert not remaining_models_to_row_count

macros/edr/materializations/model/common.sql

@@ -0,0 +1,28 @@
+{% macro query_table_metrics() %}
+  {% set query %}
+    select count(*) as row_count
+    from {{ this }}
+  {% endset %}
+
+  {% set metrics = [] %}
+  {% for metric_column in elementary.run_query(query).columns %}
+    {% set metric_name = metric_column.name %}
+    {% set metric_value = metric_column[0] %}
+    {% do metrics.append({
+      "id": "{}.{}".format(invocation_id, this),
+      "full_table_name": this | string,
+      "column_name": none,
+      "metric_name": metric_name,
+      "metric_value": metric_value 
+    }) %}
+  {% endfor %}
+  {% do return(metrics) %}
+{% endmacro %}
+
+{% macro query_metrics() %}
+  {% do return(elementary.query_table_metrics()) %}
+{% endmacro %}
+
+{% macro cache_metrics(metrics) %}
+  {% do elementary.get_cache("tables").get("metrics").extend(metrics) %}
+{% endmacro %}

macros/edr/materializations/model/incremental.sql

@@ -0,0 +1,13 @@
+{% materialization incremental, default %}
+  {% set relations = dbt.materialization_incremental_default() %}
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}
+
+{% materialization incremental, adapter="bigquery" %}
+  {% set relations = dbt.materialization_incremental_bigquery() %}
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}

macros/edr/materializations/model/table.sql

@@ -0,0 +1,13 @@
+{% materialization table, default %}
+  {% set relations = dbt.materialization_table_default() %}
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}
+
+{% materialization table, adapter="bigquery" %}
+  {% set relations = dbt.materialization_table_bigquery() %}
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}

macros/edr/materializations/model/view.sql

@@ -0,0 +1,13 @@
+{% materialization view, default %}
+  {% set relations = dbt.materialization_view_default() %}
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}
+
+{% materialization view, adapter="bigquery" %}
+  {% set relations = dbt.materialization_view_bigquery() %}
+  {% set metrics = elementary.query_metrics() %}
+  {% do elementary.cache_metrics(metrics) %}
+  {% do return(relations) %}
+{% endmaterialization %}

macros/edr/system/hooks/on_run_end.sql

@@ -5,6 +5,10 @@
         {{ return('') }}
       {% endif %}
 
+      {% if flags.WHICH in ['run', 'build'] %}
+        {{ elementary.insert_metrics() }}
+      {% endif %}
+
       {% if not elementary.get_config_var('disable_dbt_artifacts_autoupload') %}
         {{ elementary.upload_dbt_artifacts() }}
       {% endif %}

macros/edr/tests/on_run_end/insert_metrics.sql

@@ -0,0 +1,12 @@
+{% macro insert_metrics() %}
+  {% set metrics = elementary.get_cache("tables").get("metrics") %}
+  {% set database_name, schema_name = elementary.get_package_database_and_schema() %}
+  {%- set target_relation = adapter.get_relation(database=database_name, schema=schema_name, identifier='data_monitoring_metrics') -%}
+  {% if not target_relation %}
+    {% do exceptions.raise_compiler_error("Couldn't find Elementary's models. Please run `dbt run -s elementary`.") %}
+  {% endif %}
+
+  {{ elementary.file_log("Inserting metrics into {}.".format(target_relation)) }}
+  {% do elementary.insert_rows(target_relation, metrics, should_commit=true) %}
+  {{ return('') }}
+{% endmacro %}