Inviare avvisi a Slack da Snowflake

Acceda al suo account Slack e vada a questa pagina.

Clicchi su Create an App

Selezioni "From Scratch"

Nella schermata successiva assegni un nome all'app, scelga un workspace e clicchi su Create App

La schermata seguente mostrerà le credenziali dell'applicazione.

Clicchi su Incoming Webhooks nella barra laterale di sinistra:

Sposti "Active Incoming Webhooks" su "On"

Una volta cambiato l'interruttore, scorra fino in fondo alla pagina e clicchi su "Add New Webhook to Workspace".

Scelga il canale in cui far arrivare gli avvisi di Snowflake. Io ho creato un nuovo canale chiamato "snowflake-alerts" e ho aggiornato la pagina per farlo comparire nel menu a tendina.

Clicchi su Allow.

Testare il webhook

Dopo aver cliccato su Allow, la pagina successiva mostrerà l'URL del webhook e un comando CURL di test.

Copi e incolli il comando di test nel terminale. Su Windows, utilizzi Git Bash.

Il messaggio dovrebbe arrivare immediatamente sul nuovo canale Slack.

Complimenti, il nuovo webhook è operativo!

Facoltativo: assegnare all'app un avatar Snowflake ❄️ più accattivante

• Torni alla scheda "Basic Information".
• Scorra fino in fondo.
• Scarichi un logo Snowflake e lo carichi qui.
• Esegua di nuovo il test.

Ora il risultato è decisamente migliore!

L'URL del webhook contiene un secret. Sostituisca la stringa secret qui sotto con la propria ed esegua questo comando in Snowflake.

Promemoria: i secret sono oggetti a livello di schema. Presti attenzione al contesto di database e schema del worksheet, oppure usi i nomi completamente qualificati.

Nel mio caso, ho creato un secret chiamato gmds_slack_secret nello schema public del database analytics.

use schmea analytics.public;

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

Ora che il secret è stato creato, passiamo alla Notification Integration:

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');

Per inviare una notifica utilizziamo la stored procedure integrata SYSTEM$SEND_SNOWFLAKE_NOTIFICATION. È necessario passare la funzione SANITIZE_WEBHOOK_CONTENT alla procedura per rimuovere dal messaggio i placeholder (ad esempio SNOWFLAKE_WEBHOOK_SECRET).

Ecco il codice che ho eseguito sul mio account:

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');

La notifica è arrivata su Slack all'istante!

Ora che gli ingranaggi di base sono al loro posto, passiamo a un esempio concreto.

Scrivere una query SQL per individuare i picchi di utilizzo dei warehouse e inserirla in un task serverless

La query seguente confronta l'utilizzo recente (ultima ora completata) con l'utilizzo medio orario di ciascun warehouse nell'ultimo mese. In questo caso segnaleremo i warehouse con un aumento di utilizzo del 50%.

Utilizziamo un task serverless perché ci farà risparmiare! Basta omettere il nome del warehouse per rendere il task serverless.

A scopo dimostrativo, farò la union con un record fittizio così da garantire che ogni esecuzione del task produca una riga e quindi un avviso.

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

Espandi codice

Creare una tabella per memorizzare i risultati della query

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
);

Si noti che questa tabella include alcune colonne di supporto con valori di default non presenti nei risultati della query:

• alert_id: chiave primaria
• inserted_at: ci permette di sapere quando il task ha inserito il record.
• alert_sent: inizialmente false, viene impostata a true dopo l'invio dell'avviso.

Testare il task ed esaminarne i risultati

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

Creare una procedura per inviare l'avviso in caso di picco

Vogliamo ottenere quanto segue:

• Controllare la tabella usage_spike_alerts alla ricerca di record non ancora inviati: alert_sent==false
• Se ci sono record non inviati, inoltrarli a Slack.
• Contrassegnare il record come inviato.
• Mostrare nella console SQL il numero di avvisi inviati.

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
    """

Espandi codice

Ho incontrato qualche difficoltà nell'inviare a Slack i caratteri di a capo \n tramite la funzione SANITIZE_WEBHOOK_CONTENT. Slack richiede una stringa di testo letterale con doppio backslash \n, quindi una stringa Python \\n dovrebbe funzionare. Tuttavia, quando Python passa questa stringa letterale alla funzione SQL qualcosa va storto e l'avviso non viene inviato. Non sono riuscito a ottenere avvisi con a capo da Python, anche se in modalità manuale funziona: SANITIZE_WEBHOOK_CONTENT('line1\line2')

Testare manualmente la procedura

Abbiamo già inserito una riga di test nella tabella "spikes". Eseguiamo la sproc e verifichiamo che:

• i record passino a alert_sent==true
• arrivi il messaggio su Slack

execute task monitor_warehouse_spikes;  -- if you haven't already...
select * from usage_spike_alerts where not alert_sent; -- review it: sent == false
CALL send_usage_spike_alerts(); -- send the alert
-- wait for the alert to come
select * from usage_spike_alerts where not alert_sent; -- 0 rows

A questo punto dovrebbe aver ricevuto un avviso su Slack!

Concatenare il task e la procedura

Vogliamo che la schedulazione oraria del task monitor_warehouse_spikes gestisca l'intero flusso end-to-end: facciamo quindi in modo che, dopo l'esecuzione del task monitor_warehouse_spikes, venga eseguita anche la procedura che invia l'avviso.

Per prima cosa incapsuli la procedura in un task serverless e lo attivi:

CREATE TASK send_usage_spike_alerts_task
AS
CALL send_usage_spike_alerts();

alter task send_usage_spike_alerts_task resume;

Successivamente concateni i task:

ALTER TASK send_usage_spike_alerts_task
ADD AFTER monitor_warehouse_spikes;

Lo metta alla prova:

1execute task monitor_warehouse_spikes;

Il task dovrebbe ora aggiungere i nuovi record alla tabella e inviare gli avvisi a Slack!

Chiamando direttamente la procedura send_usage_spike_alerts(), il messaggio appare su Slack all'istante. Concatenando invece send_usage_spike_alerts_task al task monitor_warehouse_spikes, possono volerci fino a 3 minuti prima che l'avviso arrivi.

Esaminare il DAG dei task e lo storico

Dal menu laterale Data dell'interfaccia Snowsight (icona del database nella barra laterale), navighi fino al database e allo schema che contengono i suoi task. Espanda il contenitore dei task e selezioni uno dei due task creati. Quindi clicchi sulla scheda "Graph" nel frame principale. Da qui può attivare manualmente il task.

Può anche consultare lo storico dei task nella scheda "Run History":

La query per rilevare i picchi e l'invio dell'avviso si sarebbero potuti unire in un unico task. Si potrebbe obiettare che concatenare due task aggiunga complessità superflua. Preferisco però questa impostazione per le ragioni seguenti.

• I picchi vengono registrati in tabella in modo permanente. Tecnicamente non è indispensabile ai fini dell'alerting, ma è davvero comodo averlo.
  • Registrare i picchi in tabella è un'unità di lavoro a sé stante e merita un task dedicato.
• Separare la logica di monitoraggio dei picchi dall'avviso vero e proprio ci consente di modificare la logica senza dover ritestare la funzionalità di invio degli avvisi.

Inviare gli avvisi di Snowflake a Slack è di un'utilità incredibile. I casi d'uso per questo tipo di alerting sono infiniti!

Con questo articolo ha acquisito le competenze per:

• creare un webhook in Slack;
• creare un secret in Snowflake;
• creare una webhook integration in Snowflake;
• creare un task per registrare i picchi di utilizzo dei warehouse;
• creare una procedura per inviare avvisi al verificarsi di una condizione, in questo caso la presenza di righe restituite da una query;
• incapsulare una procedura in un task;
• concatenare più task tra loro.

Non vedo l'ora di scoprire quali casi d'uso le verranno in mente! 🥂

Jeff è un consulente Data and Analytics con oltre 15 anni di esperienza nell'automazione degli insight e nell'uso dei dati per governare i processi di business. Sul piano tecnologico è specializzato in Snowflake + dbt + Tableau. Sul fronte dei settori, vanta esperienza in Public Utility, sperimentazioni cliniche, editoria, beni di largo consumo e manifatturiero. Lo contatti pure in qualsiasi momento: [email protected].