Metron Performance Tuning Guide

Kafka Tuning

The main lever you’re going to work with when tuning Kafka throughput will be the number of partitions. A handy method for deciding how many partitions to use is to first calculate the throughput for a single producer (p) and a single consumer (c), and then use that with the desired throughput (t) to roughly estimate the number of partitions to use. You would want at least max(t/p, t/c) partitions to attain the desired throughput. See https://www.confluent.io/blog/how-to-choose-the-number-of-topicspartitions-in-a-kafka-cluster/ for more details.

Storm Tuning

There are quite a few options you will be confronted with when tuning your Storm topologies and this is largely trial and error. As a general rule of thumb, we recommend starting with the defaults and smaller numbers in terms of parallelism while iteratively working up until the desired performance is achieved. You will find the offset lag tool indispensable while verifying your settings.

We won’t go into a full discussion about Storm’s architecture - see references section for more info - but there are some general rules of thumb that should be followed. It’s first important to understand the ways you can impact parallelism in a Storm topology.

• num tasks
• num executors (parallelism hint)
• num workers

Tasks are instances of a given spout or bolt, executors are threads in a process, and workers are jvm processes. You’ll want the number of tasks as a multiple of the number of executors, the number of executors as multiple of the number of workers, and the number of workers as a multiple of the number of machines. The main reason for this approach is that it will give a uniform distribution of work to each machine and jvm process. More often than not, your number of tasks will be equal to the number of executors, which is the default in Storm. Flux does not actually provide a way to independently set number of tasks, so for enrichments and indexing which use Flux, num tasks will always equal num executors.

You can change the number of workers via the property topology.workers

Other Storm Settings

topology.max.spout.pending

This is the maximum number of tuples that can be in flight (ie, not yet acked) at any given time within your topology. You set this as a form of backpressure to ensure you don’t flood your topology.

topology.ackers.executors

This specifies how many threads should be dedicated to tuple acking. We found that setting this equal to the number of partitions in your inbound Kafka topic worked well.

spout-config.json

{
    ...
    "spout.pollTimeoutMs" : 200,
    "spout.maxUncommittedOffsets" : 10000000,
    "spout.offsetCommitPeriodMs" : 30000
}

These are the spout recommended defaults from Storm and are currently the defaults provided in the Kafka spout itself. In fact, if you find the recommended defaults work fine for you, then you can omit these settings altogether.

Performance Monitoring Tools

Before we get to tuning our cluster, it helps to describe what we might actually want to monitor as well as any potential pain points. Prior to switching over to the new Storm Kafka client, which leverages the new Kafka consumer API under the hood, offsets were stored in Zookeeper. While the broker hosts are still stored in Zookeeper, this is no longer true for the offsets which are now stored in Kafka itself. This is a configurable option, and you may switch back to Zookeeper if you choose, but Metron is currently using the new defaults. With this in mind, there are some useful tools that come with Storm and Kafka that we can use to monitor our topologies.

export BROKERLIST=<your broker comma-delimated list of host:ports>
export ZOOKEEPER=<your zookeeper comma-delimated list of host:ports>
export KAFKA_HOME=<kafka home dir>
export METRON_HOME=<your metron home>
export HDP_HOME=<your HDP home>

$ cat /tmp/consumergroup.config
security.protocol=SASL_PLAINTEXT

${KAFKA_HOME}/bin/kafka-consumer-groups.sh \
    --command-config=/tmp/consumergroup.config \
    --describe \
    --group enrichments \
    --bootstrap-server $BROKERLIST \
    --new-consumer

GROUP                          TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER
enrichments                    enrichments        9          29746066        29746067        1               consumer-2_/xxx.xxx.xxx.xxx
enrichments                    enrichments        3          29754325        29754326        1               consumer-1_/xxx.xxx.xxx.xxx
enrichments                    enrichments        43         29754331        29754332        1               consumer-6_/xxx.xxx.xxx.xxx
...

watch -n 10 -d ${KAFKA_HOME}/bin/kafka-consumer-groups.sh \
    --command-config=/tmp/consumergroup.config \
    --describe \
    --group enrichments \
    --bootstrap-server $BROKERLIST \
    --new-consumer

Parser Tuning

We’ll be using the bro sensor in this example. Note that the parsers and PCAP use a builder utility, as opposed to enrichments and indexing, which use Flux.

We started with a single partition for the inbound Kafka topics and eventually worked our way up to 48. And We’re using the following pending value, as shown below. The default is ‘null’ which would result in no limit.

storm-bro.config

{
    ...
    "topology.max.spout.pending" : 2000
    ...
}

And the following default spout settings. Again, this can be ommitted entirely since we are using the defaults.

spout-bro.config

{
    ...
    "spout.pollTimeoutMs" : 200,
    "spout.maxUncommittedOffsets" : 10000000,
    "spout.offsetCommitPeriodMs" : 30000
}

And we ran our bro parser topology with the following options. We did not need to fully match the number of Kafka partitions with our parallelism in this case, though you could certainly do so if necessary. Notice that we only needed 1 worker.

/usr/metron/0.4.0/bin/start_parser_topology.sh -k $BROKERLIST -z $ZOOKEEPER -s bro -ksp SASL_PLAINTEXT
    -ot enrichments
    -e ~metron/.storm/storm-bro.config \
    -esc ~/.storm/spout-bro.config \
    -sp 24 \
    -snt 24 \
    -nw 1 \
    -pnt 24 \
    -pp 24 \

From the usage docs, here are the options we’ve used. The full reference can be found here - https://github.com/apache/metron/blob/master/metron-platform/metron-parsers/README.md

-e,--extra_topology_options <JSON_FILE>        Extra options in the form
                                               of a JSON file with a map
                                               for content.
-esc,--extra_kafka_spout_config <JSON_FILE>    Extra spout config options
                                               in the form of a JSON file
                                               with a map for content.
                                               Possible keys are:
                                               retryDelayMaxMs,retryDelay
                                               Multiplier,retryInitialDel
                                               ayMs,stateUpdateIntervalMs
                                               ,bufferSizeBytes,fetchMaxW
                                               ait,fetchSizeBytes,maxOffs
                                               etBehind,metricsTimeBucket
                                               SizeInSecs,socketTimeoutMs
-sp,--spout_p <SPOUT_PARALLELISM_HINT>         Spout Parallelism Hint
-snt,--spout_num_tasks <NUM_TASKS>             Spout Num Tasks
-nw,--num_workers <NUM_WORKERS>                Number of Workers
-pnt,--parser_num_tasks <NUM_TASKS>            Parser Num Tasks
-pp,--parser_p <PARALLELISM_HINT>              Parser Parallelism Hint

Enrichment Tuning

We landed on the same number of partitions for enrichemnt and indexing as we did for bro - 48.

For configuring Storm, there is a flux file and properties file that we modified. Here are the settings we changed for bro in Flux. Note that the main Metron-specific option we’ve changed to accomodate the desired rate of data throughput is max cache size in the join bolts. More information on Flux can be found here - http://storm.apache.org/releases/1.0.1/flux.html

General storm settings

topology.workers: 8
topology.acker.executors: 48
topology.max.spout.pending: 2000

Spout and Bolt Settings

kafkaSpout
    parallelism=48
    session.timeout.ms=29999
    enable.auto.commit=false
    setPollTimeoutMs=200
    setMaxUncommittedOffsets=10000000
    setOffsetCommitPeriodMs=30000
enrichmentSplitBolt
    parallelism=4
enrichmentJoinBolt
    parallelism=8
    withMaxCacheSize=200000
    withMaxTimeRetain=10
threatIntelSplitBolt
    parallelism=4
threatIntelJoinBolt
    parallelism=4
    withMaxCacheSize=200000
    withMaxTimeRetain=10
outputBolt
    parallelism=48

Indexing (HDFS) Tuning

There are 48 partitions set for the indexing partition, per the enrichment exercise above.

These are the batch size settings for the bro index

cat ${METRON_HOME}/config/zookeeper/indexing/bro.json
{
  "hdfs" : {
    "index": "bro",
    "batchSize": 50,
    "enabled" : true
  }...
}

And here are the settings we used for the indexing topology

General storm settings

topology.workers: 4
topology.acker.executors: 24
topology.max.spout.pending: 2000

Spout and Bolt Settings

hdfsSyncPolicy
    org.apache.storm.hdfs.bolt.sync.CountSyncPolicy
    constructor arg=100000
hdfsRotationPolicy
    bolt.hdfs.rotation.policy.units=DAYS
    bolt.hdfs.rotation.policy.count=1
kafkaSpout
    parallelism: 24
    session.timeout.ms=29999
    enable.auto.commit=false
    setPollTimeoutMs=200
    setMaxUncommittedOffsets=10000000
    setOffsetCommitPeriodMs=30000
hdfsIndexingBolt
    parallelism: 24

PCAP Tuning

PCAP is a specialized topology that is a Spout-only topology. Both Kafka topic consumption and HDFS writing is done within a spout to avoid the additional network hop required if using an additional bolt.

General Storm topology properties

topology.workers=16
topology.ackers.executors: 0

Spout and Bolt properties

kafkaSpout
    parallelism: 128
    poll.timeout.ms=100
    offset.commit.period.ms=30000
    session.timeout.ms=39000
    max.uncommitted.offsets=200000000
    max.poll.interval.ms=10
    max.poll.records=200000
    receive.buffer.bytes=431072
    max.partition.fetch.bytes=10000000
    enable.auto.commit=false
    setMaxUncommittedOffsets=20000000
    setOffsetCommitPeriodMs=30000

writerConfig
    withNumPackets=1265625
    withMaxTimeMS=0
    withReplicationFactor=1
    withSyncEvery=80000
    withHDFSConfig
        io.file.buffer.size=1000000
        dfs.blocksize=1073741824