Writer

Flush Policies

Flushing behavior is controlled by a collection of FlushPolicy objects. They are responsible for 2 things:

1. Determining when a batch should be flushed
2. Handling a BulkWriterResponse after a batch is flushed and messages are written with a BulkMessageWriter

The FlushPolicy interface defines methods for handling these responsiblities:

• boolean shouldFlush(String sensorType, WriterConfiguration configurations, List<BulkMessage<MESSAGE_T>> messages)
• void onFlush(String sensorType, BulkWriterResponse response)

There are 2 FlushPolicy implementations included by default:

• The BatchSizePolicy will flush a batch whenever the batch size reaches a configured value. This configuration value is represented by the batchSize property in either the parser, enrichment or indexing configuration (whichever is appropriate in the current context).
• The BatchTimeoutPolicy will flush a batch whenever the batch timeout has elapsed. This configuration value is represented by the batchTimeout property in either the parser, enrichment or indexing configuration (whichever is appropriate in the current context). A maxBatchTimeout is set at creation time and serves as the ceiling for a batch timeout. In Storm topologies, this value is set to 1/2 the tuple timeout setting to ensure messages are always flushed before their tuples timeout. After a batch is flushed, the batch timer is reset for that sensor type.

For example, a configuration that sets the batchSize and batchTimeout in a parser topology will look like:

{
  "parserClassName": "org.apache.metron.parsers.bro.BasicBroParser",
  "sensorTopic": "bro",
  "parserConfig": {
    "batchSize": 5
    "batchTimeout": 2
  }
}

Similarly for the enrichment topology (configured in the Global Configuration):

{
  "enrichment.writer.batchSize": "5",
  "enrichment.writer.batchTimeout": "2",
  ...
}

And finally for the indexing topology:

{
  "elasticsearch": {
    "index": "bro",
    "batchSize": 5,
    "batchTimeout": 2,
    "enabled": true
  },
  ...
}

Additional policies can be added as needed. For example, an AckTuplesPolicy is added in the Storm bolts to handle acking tuples after a batch is flushed.

Bulk Writing Workflow

The BulkWriterComponent class collects messages in separate sensor-specific caches. This class is instantiated and supplied to classes that need to write messages to external components. A collection of default FlushPolicy implementations are created by default with the option of passing in additional FlushPolicy objects as needed.

Batching and writing messages follows this process:

1. A single message is passed to the BulkWriterComponent.write method and stored in the appropriate cache based on the sensor type. A BulkMessageWriter is also supplied to do the actual writing when messages are flushed.
2. The collection of FlushPolicy implementations are checked and a batch is flushed whenever the first FlushPolicy.shouldFlush returns true.
3. If no policies signal a flush, then nothing happens. If a policy does signal a flush, the batch of messages for that sensor are written with the supplied BulkMessageWriter. Each FlushPolicy.onFlush method is then called with the BulkWriterResponse.
4. If a sensor type has been disabled, it’s batch is flushed immediately (FlushPolicy.shouldFlush is not checked).
5. A BulkWriterComponent.flushAll method is available that immediately calls the FlushPolicy.shouldFlush methods for each sensor type in the cache. This should be called periodically by the class containing BulkWriterComponent to ensure messages are not left sitting in the cache. For example, the Storm bolts call this whenever a tick tuple is received.

Logging

Logging can be enabled for the classes described in this section to provide insight into how messages are being batched and flushed. This can be an important tool when performance tuning. Setting the log level to DEBUG on the org.apache.metron.writer package will produce detailed information about when batches are flushed, which sensor a flushed batch corresponds to, which policy caused the flush, and how long it took to write the batch.

Performance Tuning

The primary purpose of the Bulk Message Writing abstraction is to enable efficient writing to external components. This section provides recommendations and strategies for doing that. It is assumed that most Metron installations will include a variety of sensors with different volumes and velocities. Different sensors will likely need to be tuned differently.