Enviando Alertas do Snowflake para o Slack

Faça login na sua conta do Slack e acesse esta página.

Clique em Create an App

Escolha "From Scratch"

Na próxima tela, dê um nome ao app, escolha um workspace e clique em Create App

A tela seguinte vai mostrar as Application Credentials.

Clique em Incoming Webhooks na barra lateral esquerda:

Mude o toggle "Active Incoming Webhooks" para "On"

Depois de ativar, role até o fim da página e clique em "Add New Webhook to Workspace".

Escolha o canal para onde os alertas do Snowflake devem ir. Eu criei um canal novo chamado "snowflake-alerts" e atualizei a página para que ele aparecesse no dropdown.

Clique em Allow.

Teste o Webhook

Depois de clicar em Allow, a próxima página mostra a URL do seu Webhook e um comando CURL de teste.

Copie e cole o comando de teste no terminal. No Windows, use o Git Bash.

Ele deve mandar uma mensagem na hora para o seu novo canal no Slack.

Parabéns, seu Webhook novinho já está funcionando!

Opcional: dê um avatar mais bonito ao app com o floquinho do Snowflake ❄️

• Volte para a aba "Basic Information".
• Role até o final da página.
• Baixe um logotipo do Snowflake e faça o upload aqui.
• Teste de novo.

Agora sim ficou muito melhor!

A URL do Webhook contém um secret. Substitua a string de secret abaixo pela sua e rode este comando no Snowflake.

Lembrete: secrets são objetos de nível de schema. Fique atento ao contexto de database e schema da sua worksheet, ou use nomes totalmente qualificados.

No meu caso, criei um secret chamado gmds_slack_secret no schema public do database analytics.

use schmea analytics.public;

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

Agora que criamos o secret, vamos criar a 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');

Para enviar uma notificação, usamos a stored procedure nativa SYSTEM$SEND_SNOWFLAKE_NOTIFICATION. Precisamos passar a função SANITIZE_WEBHOOK_CONTENT para a procedure para remover o placeholder (ou seja, SNOWFLAKE_WEBHOOK_SECRET) da mensagem.

Veja o código que rodei na minha conta:

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

A notificação chegou no Slack na hora!

Com as peças básicas no lugar, vamos partir para um exemplo do mundo real.

Monte uma query SQL para identificar picos de uso do warehouse e empacote-a em uma serverless task

A query a seguir compara o uso recente (última hora completa) com o uso médio por hora de cada warehouse no último mês. Aqui, vamos sinalizar warehouses com aumento de 50% no uso.

Estamos usando uma serverless task porque ela vai economizar dinheiro! Basta omitir o nome do warehouse para que a task fique serverless.

Para fins de exemplo, vou fazer um union com um registro fictício para garantir que toda execução da task gere uma linha e um alerta.

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

Expandir Código

Crie uma tabela para armazenar os resultados da 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
);

Repare que essa tabela tem algumas colunas auxiliares com valores padrão que não vêm dos resultados da query:

• alert_id: chave primária
• inserted_at: ajuda a saber quando a task inseriu o registro.
• alert_sent: começa em false e passa para true depois que o alerta é enviado.

Teste a task e confira os resultados

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

Crie uma procedure para enviar o alerta quando houver um pico

Queremos fazer o seguinte:

• Checar a tabela usage_spike_alerts em busca de registros ainda não enviados: alert_sent==false
• Se houver registros pendentes, enviá-los para o Slack.
• Marcar o registro como enviado.
• Imprimir no console SQL o número de alertas enviados.

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

Expandir Código

Tive dificuldades para mandar quebras de linha \n ao Slack pela função SANITIZE_WEBHOOK_CONTENT. O Slack exige uma string literal com barra invertida dupla \n; então, em teoria, uma string Python com \\n deveria funcionar. Só que, quando o Python passa essa string literal para a função SQL, algo acontece e o alerta não sai. Não consegui fazer alertas via Python com quebras de linha, embora isso funcione manualmente: SANITIZE_WEBHOOK_CONTENT('line1\line2')

Teste a procedure manualmente

Já inserimos uma linha de teste na tabela de "spikes". Vamos rodar a sproc e garantir que:

• Os registros sejam atualizados para alert_sent==true
• A mensagem chegue no Slack

execute task monitor_warehouse_spikes;  -- se ainda não tiver feito...
select * from usage_spike_alerts where not alert_sent; -- confira: sent == false
CALL send_usage_spike_alerts(); -- dispara o alerta
-- aguarde o alerta chegar
select * from usage_spike_alerts where not alert_sent; -- 0 linhas

Você deve ter recebido um alerta no Slack!

Encadeie a task e a procedure

Queremos que o agendamento de hora em hora da task monitor_warehouse_spikes cuide do processo de ponta a ponta. Para isso, vamos garantir que, logo após rodar a task monitor_warehouse_spikes, a procedure que envia o alerta também seja executada.

Primeiro, empacote a procedure em uma serverless task e ative:

CREATE TASK send_usage_spike_alerts_task
AS
CALL send_usage_spike_alerts();

alter task send_usage_spike_alerts_task resume;

Depois, encadeie as tasks:

ALTER TASK send_usage_spike_alerts_task
ADD AFTER monitor_warehouse_spikes;

Teste:

1execute task monitor_warehouse_spikes;

Agora essa task deve adicionar os novos registros à tabela e enviar o(s) alerta(s) para o Slack!

Ao chamar a procedure send_usage_spike_alerts() diretamente, a mensagem aparece no Slack na hora. Já ao encadear a send_usage_spike_alerts_task à monitor_warehouse_spikes, o alerta pode levar até 3 minutos para chegar.

Veja o DAG da task e o histórico

No menu lateral Data da interface do Snowsight (ícone de database na barra lateral), navegue até o database e schema que contêm suas tasks. Expanda o container de tasks e selecione qualquer uma das que criamos. Em seguida, clique na aba "Graph" no quadro principal. É por aqui que você dispara a task manualmente.

Você também pode conferir o histórico da task na aba "Run History":

A query para detectar os picos e o envio do alerta poderiam ter sido feitos em uma única task. Dá para argumentar que encadear duas tasks adiciona complexidade desnecessária. Mesmo assim, gosto desse design pelos motivos a seguir.

• Os picos ficam registrados em uma tabela para sempre. Tecnicamente, isso não é necessário para o alerta, mas é muito bom ter.
  • Registrar os picos em uma tabela é uma unidade de trabalho à parte e merece sua própria task.
• Ao separar a lógica de monitoramento de picos do próprio alerta, dá para mudar a lógica sem precisar testar de novo a parte de envio.

Enviar alertas do Snowflake para o Slack é absurdamente útil. Os casos de uso para esse tipo de alerta são infinitos!

Com este artigo, você aprendeu a:

• Criar um Webhook no Slack.
• Criar um Secret no Snowflake.
• Criar uma Webhook Integration no Snowflake.
• Criar uma task para registrar picos de uso do warehouse.
• Criar uma procedure para enviar alertas com base em uma condição. Neste caso, a existência de linhas em uma query.
• Empacotar uma procedure em uma task.
• Encadear tasks.

Mal posso esperar para conhecer os casos de uso que você vai criar! 🥂

Jeff é Consultor de Dados e Analytics, com mais de 15 anos de experiência em automatizar insights e usar dados para conduzir processos de negócio. Do ponto de vista tecnológico, é especialista em Snowflake + dbt + Tableau. Já no lado de negócios, tem experiência em Serviços Públicos, Ensaios Clínicos, Editorial, CPG e Manufatura. Fale com ele quando quiser: [email protected].