Skip to content
Wombat

Pipeline

Within a Wombat configuration, in between the input and output sections, there can be an optional pipeline section containing an array of processors that are to be applied to all messages, and are not bound to any particular input or output.

If you have processors that are heavy on CPU and aren’t specific to a certain input or output they are best suited for the pipeline section. It is advantageous to use the pipeline section as it allows you to set an explicit number of parallel threads of execution:

input:
  resource: foo


pipeline:
  threads: 4
  processors:
    - mapping: |
        root = this
        fans = fans.map_each(match {
          this.obsession > 0.5 => this
          _ => deleted()
        })


output:
  resource: bar

If the field threads is set to -1 (the default) it will automatically match the number of logical CPUs available. By default almost all Wombat sources will utilise as many processing threads as have been configured, which makes horizontal scaling easy.

Metrics

The following metrics are exposed for a processor:

  • processor_received: A count of the number of messages the processor has been executed upon.
  • processor_batch_received: A count of the number of message batches the processor has been executed upon.
  • processor_sent: A count of the number of messages the processor has returned.
  • processor_batch_sent: A count of the number of message batches the processor has returned.
  • processor_error: A count of the number of times the processor has errored. In cases where an error is batch-wide the count is incremented by one, and therefore would not match the number of messages.
  • processor_latency_ns: Latency of message processing in nanoseconds. When a processor acts upon a batch of messages this latency measures the time taken to process all messages of the batch.