Logpush Transformers: filtrare e ridurre i log prima di esportarli

Il problema del volume e della sicurezza dei dati

Le organizzazioni che esportano log verso sistemi di gestione degli eventi e delle informazioni di sicurezza (SIEM), piattaforme di analisi o storage a oggetti si trovano spesso a gestire un volume di dati sproporzionato rispetto all'utilità effettiva delle informazioni. Molti log contengono record ridondanti, come i controlli di integrità (health checks) o traffico interno, che non aggiungono valore all'analisi di sicurezza ma occupano spazio e risorse. Attualmente, per rimuovere questi elementi o per omettere campi sensibili, i team devono implementare, scalare e monitorare un servizio di elaborazione separato tra l'esportazione di Cloudflare e la destinazione finale. [1]

Questa architettura non solo aumenta la complessità operativa, ma incide direttamente sui costi. Poiché i team pagano per l'esportazione, l'ingestione e l'archiviazione di ogni record, trasferire dati che verranno scartati immediatamente dopo l'arrivo a destinazione rappresenta un'inefficienza economica. Spostare la logica di filtraggio a monte, direttamente nel flusso di esportazione, permette di ridurre il volume di dati a valle, abbattendo i costi di storage e l'onere di gestione dei dati non necessari. [1]

Architettura tecnica: l'integrazione di DataFusion e Arrow

Per eliminare la necessità di servizi esterni, Cloudflare ha integrato un motore di trasformazione SQL direttamente all'interno di Bufstream, il servizio di consegna di Logpush che gestisce oltre 110 milioni di record al secondo. Inizialmente è stato valutato l'uso di WebAssembly tramite Cloudflare Workers, ma l'introduzione di una chiamata di rete avrebbe causato timeout transitori e latenze elevate, riducendo il throughput complessivo. La scelta è quindi ricaduta su Apache DataFusion, un motore di query scritto in Rust basato su Apache Arrow, che permette l'esecuzione in-process senza hop di rete. [1]

Il cuore tecnico del sistema risiede nella gestione del passaggio di dati tra il runtime Go di Bufstream e il motore Rust di DataFusion. Per evitare il costoso parsing riga per riga di formati come il JSON, viene utilizzato Arrow IPC (Inter-Process Communication), che permette di scambiare batch di dati in un formato colonnare tipizzato attraverso l'interfaccia FFI (Foreign Function Interface) e cgo. Questo approccio garantisce che DataFusion riceva i dati nel suo modello nativo, ottimizzando l'uso della CPU e mantenendo prestazioni prevedibili anche per i job ad alto volume. [1]

Nel benchmark sintetico riportato da Cloudflare, su batch di 5.000 righe risulta che l'esecuzione di un Transformer da una singola goroutine richiede circa 3 ms, il che equivale a una capacità di elaborazione di circa 1,6 milioni di righe al secondo. Per gestire i job di dimensioni massicce, Bufstream implementa un parallelismo per job, consentendo l'esecuzione concorrente dei batch di trasformazione per distribuire il lavoro di trasformazione. [1]

I due numeri descrivono livelli diversi: i record al secondo dell’intero servizio Bufstream non sono il risultato del singolo Transformer nel benchmark citato. Non vanno sommati o usati come promessa di capacità per un proprio job. Una query, il contenuto delle righe e la dimensione dei batch cambiano il lavoro necessario; la misura utile per chi adotta la funzione sarà quella del proprio flusso.

Capacità SQL e manipolazione dei record

I Transformers operano tramite una singola query SQL applicata a ogni batch di record prima della consegna. La clausola FROM deve corrispondere esattamente al dataset del job Logpush (ad esempio http_requests o audit_logs_v2), mentre la clausola WHERE definisce quali record devono essere mantenuti; i record che non soddisfano il predicato vengono scartati. L'output della SELECT diventa il nuovo record esportato in formato NDJSON. [2][1]

Il dialetto SQL supportato permette operazioni avanzate di rimodellamento. È possibile utilizzare la proiezione per rinominare campi tramite AS, creare nuovi campi calcolati, espandere array tramite UNNEST e accedere a oggetti JSON tramite l'operatore ->> (per stringhe) o -> (per oggetti). Per la creazione di strutture complesse, la funzione named_struct permette di costruire oggetti JSON annidati, mentre l'uso di parentesi quadre [value1, value2] genera array JSON. [2]

Sono inoltre disponibili funzioni scalari standard come UPPER, LOWER, COALESCE e CAST, oltre a funzioni temporali come to_timestamp ed extract per convertire i timestamp RFC3339 in formati Unix epoch. Tuttavia, sono esclusi i join, le subquery, le funzioni di aggregazione (GROUP BY, SUM, COUNT), le window functions e l'ordinamento (ORDER BY). [2]

Esempio illustrativo di implementazione

Per illustrare l'utilità pratica, consideriamo un caso in cui un'azienda voglia normalizzare i log HTTP per un'analisi multi-CDN, eliminando il rumore e proteggendo la privacy. L'obiettivo è scartare le richieste verso host interni (es. internal.example.com) e percorsi di monitoraggio (es. /healthz, /metrics), riducendo così il volume di dati inviati al SIEM. [1][2]

A livello di SELECT, la query potrebbe utilizzare regexp_replace per mascherare l'indirizzo IP del client (trasformando 203.0.113.42 in 203.0.XXX.XXX), aggiungere un campo statico 'provider' con valore 'cloudflare' e raggruppare host e path in un oggetto annidato chiamato 'request' tramite named_struct. Questo processo trasforma un record grezzo e potenzialmente sensibile in un dato normalizzato e sicuro, pronto per l'analisi senza richiedere ulteriori passaggi di pulizia a valle. [1]

Configurazione, costi e compatibilità

I Transformers sono disponibili per tutti i piani: Free, Pro, Business ed Enterprise. Un requisito fondamentale è che il job Logpush utilizzi il formato di output NDJSON; i job in formato CSV non sono compatibili e non appaiono come target per l'associazione di un transformer. La gestione può avvenire tramite il Transformer Studio nel dashboard di Cloudflare o via API utilizzando un token con permessi Logs Write. [1][2]

Dal punto di vista economico, ogni account include una quota di 1 GB di input di trasformazione al mese; l'input eccedente viene fatturato a 0,04 USD per GB. È importante notare che un job Logpush non può avere contemporaneamente sia un transformer che dei filtri classici. Se si associa un transformer a un job già filtrato, il Transformer Studio può convertire il filtro esistente in un predicato SQL da inserire nella clausola WHERE, ma l'operazione via API richiede la rimozione manuale del filtro nello stesso request di aggiornamento per evitare l'errore 400. [2]

Limiti operativi e gestione degli errori

Esistono limiti tecnici rigorosi per prevenire l'instabilità del sistema: la dimensione della query SQL non può superare i 250 KB, il nome del transformer è limitato a 255 byte e la descrizione a 4.096 byte. Inoltre, non è consentito alcun tipo di accesso al filesystem o alla rete dall'interno delle query. Ogni query viene validata contro lo schema del dataset di riferimento prima del salvataggio; errori di sintassi o riferimenti a campi inesistenti vengono rifiutati immediatamente. [2]

In caso di errore durante l'esecuzione di un batch, il sistema adotta una politica di 'fail closed': il batch non viene consegnato per evitare che dati non filtrati o non mascherati raggiungano la destinazione, violando potenzialmente le policy di sicurezza. Bufstream tenta di rieseguire il batch con ritardi progressivi; se il fallimento persiste, i record vengono eventualmente scartati e non possono essere recuperati. Gli utenti possono monitorare lo stato delle trasformazioni e i volumi di dati nella vista Transformer Health o controllare il campo last_error del job. [1][2]