Building a Fraud Detection Data Pipeline with Cribl Stream

What a detection pipeline needs

Fraud detection runs on event data. The signals you act on, purchase and card transactions, sign-on and password-reset requests, account changes, arrive buried in everything else your business logs. A working detection pipeline does three things with that stream. First, it narrows the traffic to the signals you act on: a rules engine or scoring model fed every event drowns in noise. Second, it enriches what survives the filter with the context rules need: normalized identifiers, derived labels, time buckets to group alerts by. Third, it fans the result out to more than one consumer, because detection is rarely owned by one tool. A rules engine, a correlation search index, a data-science team, and an alert archive all want the same cleaned stream.

For the fan-out, many teams put a streaming message broker in the middle: the sender publishes each filtered event once to a topic, and each downstream tool consumes the events it needs. The sections below follow the two Cribl Stream doc pages that document both halves of that pattern.

Filter and transform the events

The filter and transform page documents where custom JavaScript enters the flow. Routes carry JavaScript-enabled fields that dynamically control where data goes, and every Function has a Filter field that accepts JavaScript to determine which data it processes. Filter expressions must evaluate to either true or false. A true result means the data is processed by that Route or Function, and a false result means it passes through unchanged. The Filter field defaults to true, meaning it applies to every event by default, and you replace it with a custom expression to process only the data that matches your conditions.

The page's worked example enriches incoming data by adding a new field called phoneCompany that returns Apple or Other based on the contents of another field, using the Eval Function's Value Expression field with the expression phoneType.startsWith('iPhone') ? 'Apple' : 'Other'. Value expressions create new fields, modify existing ones, or perform transformations on data; the page's other examples convert a timestamp to an hour-based value with Math.floor(_time / 3600) and mask part of a Source string with source.replace(/.{3}/, 'XXX'). Any field that allows JavaScript also accepts Cribl Expressions, pre-built helpers for common operations that begin with C., such as C.Lookup or C.Time.

You can test an expression before committing it, in Advanced Mode. Click the Advanced mode button in the Filter expression field, paste your expression, select a sample JSON input from the drop-down or enter your own, and read the Output. For a filter expression, the output resolves to true when it is a valid expression, which means it can correctly apply the custom filter to the sample data. Lookups have one caveat: lookup files larger than 10 MB are not loaded into the browser and will not evaluate in Advanced Mode, but if the expression is valid, the updated values are still correctly applied to route and process data.

The best-practices section shows how to keep expressions predictable. In a value expression, guard against the source variable being null, undefined, or empty before reading its properties. When you are not sure a field exists, the page suggests writing (employeeID || '').length instead of employeeID.length. If a field does not exist, boolean expressions comparing its properties always evaluate to false. Wildcard lists appear in several Functions, including Eval, Mask, Publish Metrics, Parser, and Serialize. An asterisk matches one or more terms, and a leading exclamation mark negates one or more terms, so !foobar, foo* covers every term starting with foo except foobar. Lists evaluate left to right, so negations must be listed before wildcards to take precedence, and wildcards cannot target internal fields that start with two underscores; name those individually.

Fan out to detection tooling

For delivery, the Kafka destination page documents that Cribl Stream supports sending data to a Kafka topic and marks the destination's type as Streaming, the shape you want for a fan-out bus. The page states that Kafka uses a binary protocol over TCP and does not support HTTP proxies, so Cribl Stream must send events directly to receivers, and you might need to adjust firewall rules to allow the traffic.

You add the destination via QuickConnect or Routes, under a selected Worker Group. General Settings take an Output ID, a unique name for the destination definition, Bootstrap servers, a list of Kafka bootstrap servers to connect to, for example localhost:9092, and a Topic, on which events are published. The topic can be overwritten using the event's __topicOut field. Optional Settings include Record data format, which serializes events before writing to Kafka and defaults to JSON but can be set to Protobuf; Compression, where you pick None, Gzip, Snappy, LZ4, or ZSTD and Cribl strongly recommends enabling it; and Backpressure behavior, where you select whether to block, drop, or queue incoming events when all receivers exert backpressure, defaulting to Block.

The durability knob is the persistent queue. Setting backpressure behavior to Persistent Queue brings up the Persistent Queue Settings tab, and the queue buffers and preserves incoming events when a downstream Destination has an outage or experiences backpressure. The Mode option decides when the queue engages: Error when the Destination is unavailable or in an error state, Backpressure while backpressure is detected, and Always On, which queues every event to disk even without backpressure. Strict ordering is on by default, so when receivers recover, earlier queued events are sent first. On the time fields that matter for triage: by default, when an incoming event reaches the Destination, the underlying Kafka JS library adds a timestamp field set to the current time from the Date.now() method. When events arrive hours or days after they were created, the fix is to add a __kafkaTime field with an Eval Function, writing the original creation time in Unix time, seconds or milliseconds. The Destination writes that value into timestamp, and without the field it applies the default Date.now() behavior.

Delivery robustness comes from Retries: the default configuration starts with a 300 millisecond retry interval, doubles it after each retry, caps the maximum interval at 30 seconds, and attempts the request up to five times before it fails. The configuration modal also offers Live Data capture, searchable delivery Logs, a Test tab where Run Test verifies sample events send correctly, and the Monitoring page for delivery issues.

What the docs flag

Both pages carry caveats worth reading before you wire this into production:

• Leave the TLS Settings Server name (SNI) field blank on Kafka-based Sources and Destinations. The Kafka library Cribl Stream uses manages SNI without input from Cribl Stream, and setting the field in the UI can route traffic to the wrong brokers.
• Kerberos authentication is not supported for Cribl-managed Workers in Cribl.Cloud, though it is enabled for customer-managed Worker Groups, whether the Leader is based in Cribl.Cloud or on-prem.
• If you set the record data format to Protobuf, the Destination drops non-conforming events, and if it encounters a parsing error converting an event to OTLP, it discards the event and Cribl Stream logs the error.
• The persistent queue will not engage until Cribl Stream has exhausted all attempts to send events to the Kafka receiver, and multiplying the default 60 second request timeout by the default 5 retries gives a worst case of 300 seconds, or 5 minutes.
• Toggling Enable OAuth on disables the existing SASL authentication fields and replaces the SASL mechanism, authentication method, username, and password with placeholders.

Frequently asked questions

How does Cribl Stream scope which events a Function processes?

Every Function has a Filter field that accepts a JavaScript expression that must evaluate to either true or false. The field defaults to true, meaning the Function applies to every event, and you replace it with a custom expression to process only the data that matches your conditions. You can also test a filter expression against sample JSON input in Advanced Mode.

What does Cribl Stream publish to a Kafka topic, and in what format?

Events are serialized in the destination's record data format, which defaults to JSON. The topic can be overwritten per event using the event's "__topicOut" field, and by default a "timestamp" field set to the current time is added to the events. Cribl also strongly recommends enabling the compression setting.

What happens when all Kafka receivers push back on Cribl Stream?

The destination's Backpressure behavior setting chooses whether to block, drop, or queue incoming events when all receivers exert backpressure, and defaults to Block. Choosing the persistent queue buffers and preserves events while the Destination has an outage or experiences backpressure, with a Mode option controlling when the queue engages. At the default settings, the queue can take five minutes to start if sending keeps failing.

Can Cribl Stream keep the original creation time on events sent to Kafka?

It can. By default, the underlying Kafka JS library adds a "timestamp" set to the current time from the Date.now() method, but when you add a "__kafkaTime" field with the event's original creation time in Unix time using an Eval Function, the Destination writes that value into the timestamp field. If the field is not present, the Destination applies the default Date.now() behavior.