Alerts aus Snowflake an Slack senden

Melden Sie sich in Ihrem Slack-Account an und öffnen Sie diese Seite.

Klicken Sie auf "Create an App".

Wählen Sie "From Scratch".

Vergeben Sie auf dem nächsten Bildschirm einen Namen, wählen Sie einen Workspace aus und klicken Sie auf "Create App".

Auf der nächsten Seite werden Ihre Application Credentials angezeigt.

Klicken Sie in der linken Seitenleiste auf "Incoming Webhooks":

Stellen Sie "Active Incoming Webhooks" auf "On".

Scrollen Sie danach ans Ende der Seite und klicken Sie auf "Add New Webhook to Workspace".

Legen Sie fest, in welchen Channel die Snowflake-Alerts geschickt werden sollen. Ich habe dafür einen neuen Channel namens "snowflake-alerts" angelegt und die Seite anschließend neu geladen, damit er im Dropdown auftaucht.

Klicken Sie auf "Allow".

Webhook testen

Nach dem Klick auf "Allow" zeigt Ihnen die nächste Seite Ihre Webhook-URL und einen Test-CURL-Befehl.

Kopieren Sie den Testbefehl in Ihr Terminal und führen Sie ihn aus. Unter Windows nutzen Sie dafür Git Bash.

Sofort sollte in Ihrem neuen Slack-Channel eine Nachricht eintreffen.

Glückwunsch – Ihr neuer Webhook funktioniert!

Optional: Der App ein hübscheres Snowflake-Avatar ❄️ verpassen

• Wechseln Sie zurück zum Tab "Basic Information".
• Scrollen Sie ans Ende.
• Laden Sie ein Snowflake-Logo herunter und hier wieder hoch.
• Testen Sie erneut.

Schon sieht das Ganze deutlich besser aus!

Die Webhook-URL enthält ein Secret. Ersetzen Sie den Secret-String unten durch Ihren eigenen und führen Sie den Befehl in Snowflake aus.

Hinweis: Secrets sind Objekte auf Schema-Ebene. Achten Sie auf den Datenbank- und Schema-Kontext Ihres Worksheets oder verwenden Sie voll qualifizierte Namen.

Ich habe in meinem Fall ein Secret namens gmds_slack_secret im Schema public der Datenbank analytics erstellt.

use schmea analytics.public;

CREATE OR REPLACE SECRET gmds_slack_secret
  TYPE = GENERIC_STRING
  SECRET_STRING = 'T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX';

Das Secret steht – jetzt legen wir die Notification Integration an:

CREATE OR REPLACE NOTIFICATION INTEGRATION gmds_slack_webhook_integration
  TYPE=WEBHOOK
  ENABLED=TRUE
  WEBHOOK_URL='https://hooks.slack.com/services/SNOWFLAKE_WEBHOOK_SECRET'
  WEBHOOK_SECRET=analytics.public.gmds_slack_secret
  WEBHOOK_BODY_TEMPLATE='{"text": "SNOWFLAKE_WEBHOOK_MESSAGE"}'
  WEBHOOK_HEADERS=('Content-Type'='application/json');

Zum Versenden einer Benachrichtigung nutzen wir die eingebaute Stored Procedure SYSTEM$SEND_SNOWFLAKE_NOTIFICATION. Dabei müssen wir der Prozedur die Funktion SANITIZE_WEBHOOK_CONTENT übergeben, damit Platzhalter (z. B. SNOWFLAKE_WEBHOOK_SECRET) aus der Nachricht entfernt werden.

Hier der Code, den ich in meinem Account ausgeführt habe:

CREATE OR REPLACE NOTIFICATION INTEGRATION gmds_slack_webhook_integration
  TYPE=WEBHOOK
  ENABLED=TRUE
  WEBHOOK_URL='https://hooks.slack.com/services/SNOWFLAKE_WEBHOOK_SECRET'
  WEBHOOK_SECRET=analytics.public.gmds_slack_secret
  WEBHOOK_BODY_TEMPLATE='{"text": "SNOWFLAKE_WEBHOOK_MESSAGE"}'
  WEBHOOK_HEADERS=('Content-Type'='application/json');

Die Benachrichtigung kam sofort in Slack an!

Die Grundlagen stehen – weiter geht es mit einem Beispiel aus der Praxis.

SQL-Query zur Erkennung von Nutzungsspitzen schreiben und in einen Serverless Task packen

Die folgende Query vergleicht die jüngste Nutzung (letzte abgeschlossene Stunde) mit der durchschnittlichen stündlichen Nutzung jedes Warehouses im vergangenen Monat. Wir markieren dabei Warehouses, deren Nutzung um 50 % gestiegen ist.

Wir verwenden einen Serverless Task – das spart Kosten. Lassen Sie einfach den Warehouse-Namen weg, damit der Task serverless läuft.

Zu Demozwecken hänge ich per UNION einen Dummy-Datensatz an, damit jede Ausführung des Tasks eine Zeile und damit einen Alert erzeugt.

CREATE OR REPLACE TASK monitor_warehouse_spikes
SCHEDULE = 'USING CRON 2 * * * * America/New_York'
SERVERLESS_TASK_MIN_STATEMENT_SIZE = 'XSMALL'
SERVERLESS_TASK_MAX_STATEMENT_SIZE = 'XSMALL'
as
insert into usage_spike_alerts (warehouse_name,last_hour_credits,avg_monthly_credits,credit_diff,percent_increase)
WITH last_hour_usage AS (
    SELECT
        warehouse_name,
        sum(credits_used) AS last_hour_credits
    FROM
        snowflake.account_usage.warehouse_metering_history
    WHERE
        start_time >= DATEADD(hour, -2, CURRENT_TIMESTAMP)
        AND end_time <= CURRENT_TIMESTAMP

Code aufklappen

Tabelle für die Query-Ergebnisse anlegen

CREATE or replace TABLE usage_spike_alerts (
    alert_id INT AUTOINCREMENT PRIMARY KEY,
    warehouse_name STRING NOT NULL,
    last_hour_credits FLOAT NOT NULL,
    avg_monthly_credits FLOAT NOT NULL,
    credit_diff FLOAT NOT NULL,
    percent_increase FLOAT NOT NULL,
    inserted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    alert_sent boolean default false
);

Die Tabelle enthält ein paar Hilfsspalten mit Default-Werten, die nicht im Query-Ergebnis stecken:

• alert_id: Primärschlüssel
• inserted_at: zeigt, wann der Task den Datensatz eingefügt hat.
• alert_sent: anfangs false, wird nach dem Versand des Alerts auf true gesetzt.

Task testen und Ergebnisse prüfen

execute task monitor_warehouse_spikes;
select * from usage_spike_alerts where not alert_sent;

Prozedur zum Versand des Alerts bei einer Spitze erstellen

Wir möchten Folgendes erreichen:

• Die Tabelle usage_spike_alerts auf noch nicht versendete Datensätze prüfen: alert_sent==false
• Falls solche Datensätze vorliegen, an Slack schicken.
• Den Datensatz als gesendet markieren.
• Die Anzahl der versendeten Alerts in der SQL-Konsole ausgeben.

CREATE OR REPLACE PROCEDURE send_usage_spike_alerts()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'send_alerts'
AS $$
import snowflake.snowpark as snowpark

def send_alerts(session):
    query = """
        SELECT warehouse_name, last_hour_credits, avg_monthly_credits, credit_diff, percent_increase
        FROM usage_spike_alerts
        WHERE alert_sent = FALSE
    """

Code aufklappen

Beim Senden von Zeilenumbrüchen \n über die Funktion SANITIZE_WEBHOOK_CONTENT an Slack bin ich auf Probleme gestoßen. Slack erwartet ein Text-Literal mit doppeltem Backslash \n; ein Python-String \\n sollte also passen. Sobald Python dieses Literal jedoch an die SQL-Funktion übergibt, läuft irgendetwas schief und der Alert wird nicht gesendet. Python-basierte Alerts mit Zeilenumbrüchen habe ich nicht zum Laufen bekommen, manuell funktioniert das Ganze aber: SANITIZE_WEBHOOK_CONTENT('line1\line2')

Prozedur manuell testen

Wir haben bereits eine Testzeile in die "spikes"-Tabelle eingefügt. Führen wir nun die Prozedur aus und prüfen:

• Werden die Datensätze auf alert_sent==true aktualisiert?
• Kommt die Nachricht in Slack an?

execute task monitor_warehouse_spikes;  -- falls noch nicht geschehen ...
select * from usage_spike_alerts where not alert_sent; -- prüfen: sent == false
CALL send_usage_spike_alerts(); -- Alert senden
-- auf den Alert warten
select * from usage_spike_alerts where not alert_sent; -- 0 Zeilen

Sie sollten nun einen Alert in Slack bekommen haben!

Task und Prozedur verketten

Der stündliche Zeitplan des Tasks monitor_warehouse_spikes soll den kompletten Ablauf steuern. Stellen wir also sicher, dass im Anschluss an monitor_warehouse_spikes auch die Prozedur zum Versenden des Alerts ausgeführt wird.

Zuerst packen wir die Prozedur in einen Serverless Task und aktivieren ihn:

CREATE TASK send_usage_spike_alerts_task
AS
CALL send_usage_spike_alerts();

alter task send_usage_spike_alerts_task resume;

Danach verketten wir die Tasks:

ALTER TASK send_usage_spike_alerts_task
ADD AFTER monitor_warehouse_spikes;

Testen:

1execute task monitor_warehouse_spikes;

Dieser Task sollte nun die neuen Datensätze in die Tabelle einfügen und die Alerts an Slack schicken!

Beim direkten Aufruf von send_usage_spike_alerts() erscheint die Nachricht sofort in Slack. Bei der Verkettung von send_usage_spike_alerts_task mit monitor_warehouse_spikes kann es bis zu drei Minuten dauern, bis der Alert eintrifft.

Task-DAG und Verlauf prüfen

Navigieren Sie im Snowsight-UI über das Seitenmenü "Data" (Datenbank-Symbol in der Sidebar) zur Datenbank und zum Schema mit Ihren Tasks. Klappen Sie den Container "Tasks" auf und wählen Sie einen der angelegten Tasks aus. Klicken Sie im Hauptbereich auf den Tab "Graph". Hier können Sie den Task auch manuell auslösen.

Den Task-Verlauf finden Sie im Tab "Run History":

Spike-Prüfung und Alert-Versand hätten auch in einem einzigen Task laufen können. Man könnte einwenden, dass die Verkettung zweier Tasks unnötig komplex ist. Mir gefällt dieses Design aus folgenden Gründen trotzdem besser:

• Die Spikes werden dauerhaft in einer Tabelle protokolliert. Für das Alerting selbst ist das technisch nicht nötig, aber sehr praktisch.
  • Das Logging der Spikes in eine Tabelle ist eine eigenständige Arbeitseinheit und bekommt deshalb auch einen eigenen Task.
• Wenn wir die Spike-Monitoring-Logik vom eigentlichen Alert trennen, können wir die Logik anpassen, ohne den Alert-Versand erneut testen zu müssen.

Snowflake-Alerts an Slack zu schicken, ist enorm nützlich. Die Anwendungsfälle dafür sind praktisch grenzenlos!

In diesem Beitrag haben Sie gelernt, wie Sie:

• einen Webhook in Slack anlegen,
• ein Secret in Snowflake erstellen,
• eine Webhook-Integration in Snowflake einrichten,
• einen Task zum Protokollieren von Spitzen in der Warehouse-Nutzung anlegen,
• eine Prozedur erstellen, die Alerts auf Basis einer Bedingung verschickt – hier am Beispiel vorhandener Zeilen in einer Query,
• eine Prozedur in einen Task verpacken,
• Tasks miteinander verketten.

Ich bin gespannt, welche Anwendungsfälle Ihnen dazu einfallen! 🥂

Jeff ist Data- und Analytics-Consultant mit über 15 Jahren Erfahrung darin, Insights zu automatisieren und Geschäftsprozesse datenbasiert zu steuern. Technisch ist er auf Snowflake + dbt + Tableau spezialisiert. Fachlich hat er Erfahrung in den Bereichen Versorgungswirtschaft, klinische Studien, Verlagswesen, CPG und Fertigung. Schreiben Sie ihm jederzeit gerne unter [email protected].