Creazione e consumo di webhook per uccelli
Uccello
27 gen 2022
1 min read
Conclusioni principali
I webhook degli eventi in tempo reale di Bird consentono ai mittenti di ricevere i dati degli eventi istantaneamente—senza polling, senza lavori cron e senza problemi di limite di velocità.
I webhook eliminano la complessità della gestione delle finestre temporali, prevenendo eventi mancati e gestendo record duplicati.
Ideale per l'automazione a valle: aggiornare elenchi, iniziare percorsi, arricchire dashboard o sincronizzare sistemi interni.
Il tutorial guida i mittenti nella costruzione di una pipeline di ingestione AWS completa: archiviazione S3, elaborazione Lambda e un load balancer applicativo.
S3 serve come livello di archiviazione centrale per i payload dei webhook, con ciascun batch scritto come file JSON piatto.
Le funzioni Lambda gestiscono sia l'ingestione (memorizzazione di batch grezzi) che la trasformazione (conversione da JSON a CSV).
Bird suddivide gli eventi—in ogni batch di webhook è incluso un
x-messagesystems-batch-idunivoco che consente controlli di replay e deduplicazione.Il Lambda consumer deve rimanere efficiente poiché Bird ritenta i batch se l'endpoint non risponde entro circa 10 secondi.
Si consiglia AWS ALB (rispetto a API Gateway) per la semplicità, l'efficacia dei costi e la gestione diretta delle richieste HTTP POST.
DNS (Route 53 o fornitore esterno) è configurato per mappare un hostname amichevole all'endpoint ALB.
Dopo la configurazione, Bird invia i dati degli eventi direttamente e in modo affidabile alla tua pipeline AWS per l'elaborazione asincrona.
La guida copre anche le migliori pratiche: autorizzazioni IAM del minimo privilegio, vincoli di archiviazione temporanea, evitare trigger ricorsivi e organizzare flussi di lavoro multi-lambda.
Q&A Highlights
Qual è lo scopo principale dei webhooks degli eventi in tempo reale di Bird?
Per inviare i dati degli eventi direttamente all'endpoint di un mittente in tempo reale, abilitando l'automazione immediata senza interrogazioni o chiamate API con limiti di frequenza.
Perché i webhooks sono migliori rispetto al recupero dei dati tramite API per i grandi mittenti?
Poiché le API pull richiedono la gestione delle finestre temporali, rischiano lacune o duplicati nei dati e possono superare i limiti di velocità, i webhook eliminano tutto ciò inviando dati in modo continuo.
Quali servizi AWS vengono utilizzati nella pipeline di acquisizione webhook raccomandata?
Amazon S3 (storage), AWS Lambda (elaborazione), un Application Load Balancer (listener HTTP) e opzionalmente Route 53 (DNS).
Come fa Bird a raggruppare i dati degli eventi?
Bird invia più eventi insieme in un payload, ognuno assegnato un ID di batch unico (x-messagesystems-batch-id) per tracciamento, tentativi di nuovo invio e deduplicazione.
Cosa attiva la funzione Lambda del consumatore?
L'ALB inoltra le richieste POST di webhook in arrivo direttamente alla Lambda, che estrae il payload e lo scrive su S3.
Perché memorizzare il batch webhook grezzo in S3 prima di elaborarlo?
Per garantire un'ingestione rapida (<10 secondi) in modo che la connessione non scada, e per scaricare l'elaborazione più pesante su un pipeline asincrona separata.
Cosa fa la seconda funzione Lambda?
Viene attivato da nuovi oggetti S3, valida il JSON, lo converte in CSV e scrive l'output elaborato in un bucket S3 separato.
Perché usare un bucket S3 separato per i file CSV elaborati?
Per evitare attivazioni ricorsive (scrivere un nuovo file processato nello stesso bucket innescherebbe nuovamente la Lambda all'infinito).
Quali permessi richiedono le funzioni Lambda?
Il consumer Lambda ha bisogno delle autorizzazioni S3 PutObject; il processing Lambda ha bisogno di GetObject per il bucket di origine e PutObject per il bucket di destinazione.
Perché è consigliato un AWS ALB rispetto ad API Gateway?
ALBs sono più economici, più semplici e ideali per un inoltro HTTPS POST diretto; API Gateway può modificare il formato del payload e aumentare la complessità.
Come configurano i mittenti il webhook in Bird?
Fornendo l'endpoint HTTPS (il record DNS per l'ALB), selezionando un subaccount, scegliendo gli eventi e salvando la configurazione del webhook.
Quali opzioni downstream esistono per archiviare o analizzare i dati elaborati?
Carica nei database (PostgreSQL, DynamoDB, RDS), immetti nei sistemi ETL o esegui query direttamente con strumenti come Athena.
Di seguito è riportata una semplice guida per aiutare i mittenti a sentirsi a proprio agio quando creano un webhook per eventi Bird e consumano i dati utilizzando l'infrastruttura di AWS.
I webhook degli eventi in tempo reale di Bird sono uno strumento incredibilmente prezioso per i mittenti per avere i dati automaticamente inviati ai loro sistemi. Questo può guidare l'automazione a valle come l'aggiornamento delle liste di distribuzione, l'attivazione di percorsi e-mail automatizzati o il popolamento di dashboard interni. Mentre gli stessi dati degli eventi possono essere accessibili tramite l'interfaccia utente Bird utilizzando la ricerca eventi o in modo programmatico sfruttando la Events API di Bird, le limitazioni sul numero di record restituiti in una singola richiesta o i limiti di frequenza imposti sull'endpoint API possono rendere entrambi questi metodi restrittivi per i mittenti grandi e sofisticati.
I webhook degli eventi in tempo reale permettono a un mittente di configurare un endpoint al quale Bird trasmette i dati, e i dati possono essere consumati senza dover programmare cron job che estraggano i dati. Ci sono anche compromessi logistici quando si estraggono i dati invece di averli inviati, come dover identificare quale periodo di tempo e parametri utilizzare per ciascuna richiesta API. Se i periodi di tempo non sono perfettamente allineati, si rischia di perdere dati, e se i periodi si sovrappongono, è necessario gestire i record di dati duplicati. Con i webhook in tempo reale, i dati degli eventi vengono semplicemente inviati al tuo endpoint non appena diventano disponibili all'interno di Bird.
Mentre i benefici di ricevere dati degli eventi in tempo reale per guidare i processi di automazione a valle possono essere immediatamente compresi da molti mittenti, il processo effettivo per implementare e consumare i webhook può essere intimidatorio. Questo può essere particolarmente vero se non si ha familiarità con i componenti tecnici della creazione di un endpoint e della gestione dei dati in modo programmatico. Ci sono servizi disponibili che consumeranno i dati webhook di Bird e li ETL nel tuo database automaticamente – un esempio sarebbe StitchData, di cui abbiamo parlato in un blog in passato. Tuttavia, se desideri più controllo sul processo, puoi facilmente costruire i componenti da solo. Di seguito è riportata una guida semplice per aiutare i mittenti a sentirsi a loro agio quando creano un webhook degli eventi Bird e consumano i dati utilizzando l'infrastruttura all'interno di AWS.
Configurazione dell'Endpoint di Destinazione Webhook
Quando viene creato un evento Bird, vogliamo che i dati dell'evento siano trasmessi in tempo reale a un endpoint su AWS in modo da poter consumare e utilizzare tali dati in modo programmatico. I dati verranno inviati da Bird a un endpoint di destinazione, che inoltrerà il payload a una funzione lambda che elaborerà e memorizzerà i dati in un bucket S3. Di seguito è possibile vedere un diagramma di alto livello del flusso di dati descritto:
Per implementare questo flusso di lavoro, costruiamoli effettivamente in ordine inverso a partire dalla creazione di un bucket S3 in cui memorizzeremo i nostri dati degli eventi e quindi lavoriamo a ritroso, aggiungendo ogni componente che alimenta ciò che abbiamo costruito.
Creare un S3 Bucket per memorizzare i dati del Webhook
Crea una funzione Lambda per consumare i dati
Il reale elaborazione e memorizzazione dei dati sarà eseguito da una lambda function che viene invocata dal nostro application load balancer (ALB).
Il primo passo è creare la tua lambda function navigando al servizio Lambda all'interno di AWS e cliccando su "Crea Function". Ti verrà richiesto di assegnare un nome alla tua lambda function e selezionare quale linguaggio di programmazione utilizzare per scrivere la tua funzione. Per questo esempio, usiamo Python come linguaggio di runtime.
Ora dobbiamo sviluppare la nostra lambda function. Per un momento, assumiamo che il nostro application load balancer sia stato configurato e stia inoltrando il payload del webhook alla nostra lambda function – la lambda riceverà un payload che include l'intera intestazione e corpo. Il payload viene passato alla nostra lambda function utilizzando l'oggetto "event" come un dizionario. Puoi fare riferimento all'intestazioni e al corpo del payload indipendentemente accedendo agli oggetti "headers" e "body" all'interno del payload. In questo esempio leggeremo semplicemente l'intestazione "x-messagesystems-batch-id", dove l'ID batch è un valore unico creato da Bird per il batch del webhook, e lo useremo come nome del file quando memorizzeremo il corpo come file piatto in S3; tuttavia, potresti voler aggiungere funzionalità aggiuntive come controlli di autenticazione o gestione degli errori, secondo necessità.
Quando si memorizza il payload in un file piatto su S3, dovremo definire il nome del bucket S3, la posizione e il nome del file in cui i dati del payload saranno memorizzati. Nella nostra funzione lambda di esempio, facciamo questo all'interno della funzione "store_batch". In questo esempio, memorizzeremo l'intero batch come un singolo file, il che aiuta a garantire che i dati vengano raccolti e memorizzati prima che la connessione HTTP tra Bird e il tuo endpoint scada. Mentre potresti regolare le impostazioni di timeout della connessione sul tuo load balancer, non ci sono garanzie che la connessione non scada dal lato della trasmissione (in questo caso Bird) o che la connessione non venga terminata prima che la tua lambda function finisca di eseguire. È una buona pratica mantenere la tua consumer function il più efficiente possibile e riservare le attività di elaborazione dei dati per processi a valle dove possibile – come convertire il payload formato JSON in batch in un file CSV, o caricare i dati dell'evento in un database.
È importante notare che potrebbe essere necessario aggiornare le autorizzazioni per la tua funzione lambda. Il tuo ruolo di esecuzione avrà bisogno delle autorizzazioni PutObject e GetObject per S3. È una buona pratica applicare il principio del minimo privilegio, quindi consiglio di impostare queste autorizzazioni solo per il bucket S3 dove i payload del webhook saranno memorizzati.
Un esempio della nostra funzione lambda consumer può essere trovato qui.
Una nota veloce sull'ID batch: Bird compilerà eventi in un singolo payload, dove ogni batch può contenere da 1 a 350 o più record di eventi. Il batch riceverà un ID batch unico, che può essere utilizzato per visualizzare lo stato del batch sfruttando l'Event Webhooks API o all'interno del tuo account Bird cliccando su un webhook stream e selezionando "Batch Status". Nel caso in cui un payload del webhook non possa essere consegnato, come durante un timeout della connessione, Bird automaticamente ritenterà il batch utilizzando lo stesso ID batch. Questo può accadere quando la tua lambda function sta operando vicino al tempo massimo di round-trip di 10 secondi ed è un motivo per ottimizzare la consumer function per ridurre il tempo di esecuzione.
Per gestire tutte le attività di elaborazione dei dati, consiglio di creare una funzione lambda separata che si esegue ogni volta che viene creato un nuovo file sul bucket S3 – in questo modo, l'elaborazione dei dati viene eseguita asincronamente alla trasmissione dei dati, e non vi è alcun rischio di perdere dati a causa di una connessione terminata. Discuto la funzione lambda di elaborazione in una sezione successiva.
Crea un Application Load Balancer
Per ricevere un webhook payload, dobbiamo fornire un endpoint a cui inviare i payload. Facciamo questo creando un application load balancer all'interno di AWS navigando verso EC2 > Load Balancers e cliccando su "Create Load Balancer." Ti verrà chiesto di scegliere quale tipo di load balancer vuoi creare – per questo, vogliamo creare un application load balancer. Dobbiamo usare un application load balancer (ALB) per costruire il nostro consumer perché gli eventi webhooks verranno inviati come richieste HTTP, e gli ALB vengono utilizzati per il routing delle richieste HTTP all'interno di AWS. Potremmo implementare un HTTP Gateway come alternativa; tuttavia, utilizziamo un ALB per questo progetto perché è più leggero ed economico rispetto a HTTP Gateway. È importante notare che se scegli di utilizzare un HTTP Gateway, il formato degli eventi potrebbe essere diverso rispetto a un ALB, e quindi la tua funzione lambda dovrà gestire correttamente l'oggetto della richiesta.
Una volta creato il tuo ALB, ti verrà chiesto di assegnare un nome al tuo ALB e configurare lo schema e le impostazioni di accesso/sicurezza – poiché prevediamo di ricevere dati di eventi da una fonte esterna (Bird), vogliamo che il nostro ALB sia internet-facing. Sotto "Listeners and routing," l'ALB dovrebbe ascoltare HTTPS sulla porta 443, e vogliamo creare un Target group che punti alla nostra funzione lambda in modo che il nostro ALB inoltri le richieste in entrata alla funzione lambda consume che abbiamo creato sopra. Devi inoltre assicurarti che il security group abbia il permesso di accettare il traffico attraverso la porta 443.
Crea un Record DNS per il Load Balancer
Per semplificare l'uso del nostro ALB come endpoint, creeremo un record A nel DNS che punta al nostro ALB. Per questo, possiamo usare il servizio AWS Route 53 (o il tuo attuale provider DNS) e creare un record A per il nome host che desideri utilizzare per il tuo endpoint (ad es. spevents.<your_domain>). Quando si lavora con DNS su larga scala su AWS, tieni presente che ci sono limiti DNS non documentati che possono influenzare applicazioni ad alto volume, specialmente quelle che gestiscono grandi quantità di traffico in uscita come i sistemi di consegna email. Il record A dovrebbe essere configurato per puntare all'ALB che abbiamo creato. Se stai usando Route 53 per gestire i record DNS, puoi fare riferimento all'istanza ALB direttamente abilitando "Alias" e selezionando l'ALB; altrimenti, se stai usando un provider DNS esterno, dovresti puntare il record A all'indirizzo IP pubblico dell'istanza ALB.
Consiglio di usare uno strumento come Postman per testare che tutto sia stato configurato correttamente prima di abilitare il tuo Bird webhook. Puoi fare una richiesta POST al tuo endpoint e confermare che viene ricevuta una risposta. Se la tua richiesta POST non restituisce una risposta, potrebbe essere necessario verificare che il tuo ALB stia ascoltando sulla porta corretta.
Crea un Bird Webhook
Ora siamo pronti per creare il webhook in Bird e utilizzare il nome host definito dal record A sopra come nostro endpoint di destinazione. Per creare il webhook, naviga alla sezione Webhooks all'interno del tuo account Bird e clicca su "Create Webhook". Ti verrà chiesto di assegnare un nome al tuo webhook e fornire un URL di destinazione - la destinazione dovrebbe essere il nome host del record A che hai creato in precedenza. Nota che l'URL di destinazione potrebbe richiedere l'inclusione di "HTTPS://" nell'URL.
Una volta completato, verifica che il subaccount corretto e gli eventi siano selezionati, e premi "Create Webhook" per salvare la tua configurazione. I dati dell'evento per tutti i tipi di evento selezionati ora verranno trasmessi al nostro URL di destinazione e consumati dal nostro ALB per l'elaborazione a valle.
Elaborazione Dati Evento Webhook
In base all'uso previsto per l'archiviazione dei dati degli eventi Bird, i tuoi requisiti potrebbero essere soddisfatti semplicemente archiviando il payload JSON come un file piatto. Potresti anche avere un processo ETL downstream già stabilito in grado di consumare e caricare dati in formato JSON. In entrambi questi casi, potresti essere in grado di utilizzare il file piatto creato dal nostro lambda di elaborazione che abbiamo creato sopra così com'è.
In alternativa, potresti aver bisogno di trasformare i dati – ad esempio per convertire da un formato JSON a un CSV – o caricare i dati direttamente in un database. In questo esempio, creeremo una semplice funzione lambda che convertirà i dati del webhook dal formato JSON originale in un file CSV che potrebbe essere caricato in un database.
Crea un Lambda per Processare i Dati
Come con la funzione lambda per consumare i dati del webhook, dobbiamo creare una nuova funzione lambda navigando nel servizio Lambda all'interno di AWS e premendo "Create Function". Questa nuova funzione lambda verrà attivata quando viene creato un nuovo file nel nostro bucket S3: leggerà i dati e li convertirà in un nuovo file CSV.
La funzione lambda accetta le informazioni del file come un evento. Nella funzione lambda di esempio, vedrai che abbiamo prima una serie di controlli di validità per garantire che i dati siano completi e formattati come previsto. Successivamente, convertiamo il payload JSON in un file CSV utilizzando la libreria "csv" e scrivendo in un file temporaneo. Le funzioni lambda possono scrivere i file locali solo nella directory "/tmp", quindi creiamo un file CSV temporaneo e lo chiamiamo con la convenzione <batch_id>.csv. Il motivo per cui utilizziamo il batch_id qui è semplicemente per garantire che eventuali processi paralleli in esecuzione a seguito della ricezione di più payload webhook non interferiscano tra loro, poiché ogni batch webhook avrà un batch_id univoco.
Una volta che i dati sono stati completamente convertiti in CSV, leggiamo i dati CSV come un flusso di byte, cancelliamo il file temporaneo e salviamo i dati CSV come un nuovo file su S3. È importante notare che è necessario un bucket S3 diverso per l'output, altrimenti rischiamo di creare un loop ricorsivo che può portare a un aumento dell'uso di lambda e a costi maggiori. Dovremo identificare in quale bucket e posizione S3 vogliamo che il nostro file CSV venga memorizzato all'interno della nostra funzione lambda. Seguire la stessa procedura sopra per creare un nuovo bucket S3 per memorizzare il nostro file CSV.
Notare che la directory tmp è limitata a 512 MB di spazio, quindi è importante che il file temporaneo venga eliminato successivamente per garantire un sufficiente spazio per le esecuzioni future. Il motivo per cui utilizziamo un file temporaneo, anziché scrivere direttamente su S3, è semplificare la connessione a S3 con una singola richiesta.
Proprio come con la funzione lambda di consumo, potrebbe essere necessario aggiornare i permessi per la tua funzione lambda di processo. Questa funzione lambda richiede che il ruolo di esecuzione abbia i permessi GetObject per il bucket S3 di input, e sia PutObject che GetObject per il bucket S3 di output.
Un esempio della nostra funzione lambda di elaborazione può essere trovato qui.
Configura una Lambda per l'esecuzione quando nuovi dati vengono archiviati su S3
Ora che la nostra funzione lambda per convertire il file da formato JSON a CSV è stata creata, dobbiamo configurarla per attivarsi quando viene creato un nuovo file nel nostro bucket S3. Per fare questo, dobbiamo aggiungere un trigger alla nostra funzione lambda aprendo la nostra funzione lambda e cliccando su "Add Trigger" in cima alla pagina. Seleziona "S3" e fornisci il nome del bucket S3 dove sono archiviati i payload raw del webhook. Hai anche l'opzione di specificare un prefisso e/o suffisso file su cui filtrare. Una volta configurate le impostazioni, puoi aggiungere il trigger cliccando su "Add" in fondo alla pagina. Ora la tua funzione lambda di elaborazione si eseguirà ogni volta che un nuovo file viene aggiunto al tuo bucket S3.
Caricamento dei dati in un Database
In questo esempio, non coprirò in dettaglio il caricamento dei dati in un database, ma se hai seguito questo esempio hai un paio di opzioni:
Carica i dati direttamente nel tuo database all'interno della tua funzione lambda di elaborazione
Consuma il tuo file CSV utilizzando un processo ETL consolidato
Sia che tu stia utilizzando un servizio di database AWS, come RDS o DynamoDB, o se hai il tuo database PostgreSQL (o simile), puoi connetterti al tuo servizio di database direttamente dalla tua funzione lambda di processo. Ad esempio, nello stesso modo in cui abbiamo chiamato il servizio S3 utilizzando “boto3” nella nostra funzione lambda, potresti anche utilizzare “boto3” per chiamare RDS o DynamoDB. Il servizio AWS Athena potrebbe essere utilizzato anche per leggere i file di dati direttamente dai flat-file e accedere ai dati utilizzando un linguaggio di query simile a SQL. Consiglio di fare riferimento alla documentazione rispettiva del servizio che stai utilizzando per maggiori informazioni su come meglio realizzare questo all'interno del tuo ambiente.
Analogamente, ci sono molti servizi disponibili che possono aiutare a consumare file CSV e caricare i dati in un database. Potresti già avere un processo ETL consolidato che puoi sfruttare.
Speriamo che questa guida ti sia stata utile – buon invio!
