Contents

• Introduction
• Parser Error Routing
• Filtering
• Parser Architecture
• Message Format
• Global Configuration
• Parser Configuration
• Parser Adapters
• Kafka Queue
• JSON Path

Introduction

Parsers are pluggable components which are used to transform raw data (textual or raw bytes) into JSON messages suitable for downstream enrichment and indexing.

There are two general types types of parsers:

• A parser written in Java which conforms to the MessageParser interface. This kind of parser is optimized for speed and performance and is built for use with higher velocity topologies. These parsers are not easily modifiable and in order to make changes to them the entire topology need to be recompiled.
• A general purpose parser. This type of parser is primarily designed for lower-velocity topologies or for quickly standing up a parser for a new telemetry before a permanent Java parser can be written for it. As of the time of this writing, we have:
  • Grok parser: org.apache.metron.parsers.GrokParser with possible parserConfig entries of
    • grokPath : The path in HDFS (or in the Jar) to the grok statement. By default attempts to load from HDFS, then falls back to the classpath, and finally throws an exception if unable to load a pattern.
    • patternLabel : The pattern label to use from the grok statement
    • multiLine : The raw data passed in should be handled as a long with multiple lines, with each line to be parsed separately. This setting’s valid values are ‘true’ or ‘false’. The default if unset is ‘false’. When set the parser will handle multiple lines with successfully processed lines emitted normally, and lines with errors sent to the error topic.
    • timestampField : The field to use for timestamp. If your data does not have a field exactly named “timestamp” this field is required, otherwise the record will not pass validation. If the timestampField is also included in the list of timeFields, it will first be parsed using the provided dateFormat.
    • timeFields : A list of fields to be treated as time.
    • dateFormat : The date format to use to parse the time fields. Default is “yyyy-MM-dd HH:mm:ss.S z”.
    • timezone : The timezone to use. UTC is default.
    • The Grok parser supports either 1 line to parse per incoming message, or incoming messages with multiple log lines, and will produce a json message per line
  • CSV Parser: org.apache.metron.parsers.csv.CSVParser with possible parserConfig entries of
    • timestampFormat : The date format of the timestamp to use. If unspecified, the parser assumes the timestamp is ms since unix epoch.
    • columns : A map of column names you wish to extract from the CSV to their offsets (e.g. { 'name' : 1, 'profession' : 3} would be a column map for extracting the 2nd and 4th columns from a CSV)
    • separator : The column separator, , by default.
  • JSON Map Parser: org.apache.metron.parsers.json.JSONMapParser with possible parserConfig entries of
    • mapStrategy : A strategy to indicate how to handle multi-dimensional Maps. This is one of
      • DROP : Drop fields which contain maps
      • UNFOLD : Unfold inner maps. So { "foo" : { "bar" : 1} } would turn into {"foo.bar" : 1}
      • ALLOW : Allow multidimensional maps
      • ERROR : Throw an error when a multidimensional map is encountered
    • jsonpQuery : A JSON Path query string. If present, the result of the JSON Path query should be a list of messages. This is useful if you have a JSON document which contains a list or array of messages embedded in it, and you do not have another means of splitting the message.
    • wrapInEntityArray : "true" or "false". If jsonQuery is present and this flag is present and set to "true", the incoming message will be wrapped in a JSON entity and array. for example: {"name":"value"},{"name2","value2} will be wrapped as {"message" : [{"name":"value"},{"name2","value2}]}. This is using the default value for wrapEntityName if that property is not set.
    • wrapEntityName : Sets the name to use when wrapping JSON using wrapInEntityArray. The jsonpQuery should reference this name.
    • A field called timestamp is expected to exist and, if it does not, then current time is inserted.
  • Regular Expressions Parser
    • recordTypeRegex : A regular expression to uniquely identify a record type.
    • messageHeaderRegex : A regular expression used to extract fields from a message part which is common across all the messages.
    • convertCamelCaseToUnderScore : If this property is set to true, this parser will automatically convert all the camel case property names to underscore seperated. For example, following conversions will automatically happen:

      ipSrcAddr -> ip_src_addr
      ipDstAddr -> ip_dst_addr
      ipSrcPort -> ip_src_port
      

      Note this property may be necessary, because java does not support underscores in the named group names. So in case your property naming conventions requires underscores in property names, use this property.

    • fields : A json list of maps contaning a record type to regular expression mapping.

    A complete configuration example would look like:

    "convertCamelCaseToUnderScore": true,
    "recordTypeRegex": "kernel|syslog",
    "messageHeaderRegex": "(<syslogPriority>(<=^&lt;)\\d{1,4}(?=>)).*?(<timestamp>(<=>)[A-Za-z] {3}\\s{1,2}\\d{1,2}\\s\\d{1,2}:\\d{1,2}:\\d{1,2}(?=\\s)).*?(<syslogHost>(<=\\s).*?(?=\\s))",
    "fields": [
      {
        "recordType": "kernel",
        "regex": ".*(<eventInfo>(<=\\]|\\w\\:).*?(?=$))"
      },
      {
        "recordType": "syslog",
        "regex": ".*(<processid>(<=PID\\s=\\s).*?(?=\\sLine)).*(<filePath>(<=64\\s)\/([A-Za-z0-9_-]+\/)+(?=\\w))        (<fileName>.*?(?=\")).*(<eventInfo>(<=\").*?(?=$))"
      }
    ]
    

    Note: messageHeaderRegex and regex (withing fields) could be specified as lists also e.g.

    "messageHeaderRegex": [
      "regular expression 1",
      "regular expression 2"
    ]
    

    Where regular expression 1 are valid regular expressions and may have named groups, which would be extracted into fields. This list will be evaluated in order until a matching regular expression is found.

    messageHeaderRegex is run on all the messages. Yes, all the messages are expected to contain the fields which are being extracted using the messageHeaderRegex. messageHeaderRegex is a sort of HCF (highest common factor) in all messages.

    recordTypeRegex can be a more advanced regular expression containing named goups. For example

    “recordTypeRegex”: “(<process>(<=\s)\b(kernel|syslog)\b(?=\[|:))”

    Here all the named goups (process in above example) will be extracted as fields.

    Though having named group in recordType is completely optional, still one could want extract named groups in recordType for following reasons:

    1. Since recordType regular expression is already getting matched and we are paying the price for a regular expression match already, we can extract certain fields as a by product of this match.
    2. Most likely the recordType field is common across all the messages. Hence having it extracted in the recordType (or messageHeaderRegex) would reduce the overall complexity of regular expressions in the regex field.

    regex within a field could be a list of regular expressions also. In this case all regular expressions in the list will be attempted to match until a match is found. Once a full match is found remaining regular expressions are ignored.

    "regex":  [ "record type specific regular expression 1",
                "record type specific regular expression 2"]
    

    timesamp

    Since this parser is a general purpose parser, it will populate the timestamp field with current UTC timestamp. Actual timestamp value can be overridden later using stellar. For example in case of syslog timestamps, one could use following stellar construct to override the timestamp value. Let us say you parsed actual timestamp from the raw log:

    <38>Jun 20 15:01:17 hostName sshd[11672]: Accepted publickey for prod from 55.55.55.55 port 66666 ssh2

    syslogTimestamp=“Jun 20 15:01:17”

    Then something like below could be used to override the timestamp.

    "timestamp_str": "FORMAT('%s%s%s', YEAR(),' ',syslogTimestamp)",
    "timestamp":"TO_EPOCH_TIMESTAMP(timestamp_str, 'yyyy MMM dd HH:mm:ss' )"
    

    OR, if you want to factor in the timezone

    "timestamp":"TO_EPOCH_TIMESTAMP(timestamp_str, timestamp_format, timezone_name )"
    

Parser Message Routing

Messages are routed to the Kafka enrichments topic by default. The output topic can be changed with the output_topic option when Starting the Parser Topology or with the outputTopic Parser Configuration setting. The order of precedence from highest to lowest is as follows:

1. Parser start script option
2. Parser configuration setting
3. Default enrichments topic

A message can also be routed to other locations besides Kafka with the writerClassName Parser Configuration setting. Messages can be routed independently for each sensor type when configured with Parser Configuration settings.

Parser Error Routing

Currently, we have a few mechanisms for either deferring processing of messages or marking messages as invalid.

Filtering

One can also filter a message by specifying a filterClassName in the parser config. Filtered messages are just dropped rather than passed through.

Parser Architecture

Data flows through the parser via kafka and into the enrichments topology in kafka. Errors are collected with the context of the error (e.g. stacktrace) and original message causing the error and sent to an error queue. Invalid messages as determined by global validation functions are also treated as errors and sent to an error queue.

Message Format

All Metron messages follow a specific format in order to ingest a message. If a message does not conform to this format it will be dropped and put onto an error queue for further examination. The message must be of a JSON format and must have a JSON tag message like so:

{"message" : message content}

Where appropriate there is also a standardization around the 5-tuple JSON fields. This is done so the topology correlation engine further down stream can correlate messages from different topologies by these fields. We are currently working on expanding the message standardization beyond these fields, but this feature is not yet availabe. The standard field names are as follows:

• ip_src_addr: layer 3 source IP
• ip_dst_addr: layer 3 dest IP
• ip_src_port: layer 4 source port
• ip_dst_port: layer 4 dest port
• protocol: layer 4 protocol
• timestamp (epoch)
• original_string: A human friendly string representation of the message

The timestamp and original_string fields are mandatory. The remaining standard fields are optional. If any of the optional fields are not applicable then the field should be left out of the JSON.

So putting it all together a typical Metron message with all 5-tuple fields present would look like the following:

{
  "message": {
    "ip_src_addr": xxxx,
    "ip_dst_addr": xxxx,
    "ip_src_port": xxxx,
    "ip_dst_port": xxxx,
    "protocol": xxxx,
    "original_string": xxx,
    "additional-field 1": xxx
  }
}

Global Configuration

There are a few properties which can be managed in the global configuration that have pertinence to parsers and parsing in general.

Parser Configuration

The configuration for the various parser topologies is defined by JSON documents stored in zookeeper.

The document is structured in the following way

• parserClassName : The fully qualified classname for the parser to be used.
• filterClassName : The filter to use. This may be a fully qualified classname of a Class that implements the org.apache.metron.parsers.interfaces.MessageFilter<JSONObject> interface. Message Filters are intended to allow the user to ignore a set of messages via custom logic. The existing implementations are:
  • STELLAR : Allows you to apply a stellar statement which returns a boolean, which will pass every message for which the statement returns true. The Stellar statement that is to be applied is specified by the filter.query property in the parserConfig.

    Example Stellar Filter which includes messages which contain a the field1 field:

    {
      "filterClassName" : "STELLAR",
      "parserConfig" : {
        "filter.query" : "exists(field1)"
      }
    }
    

• writerClassName : The class used to write messages after they have been parsed. Defaults to org.apache.metron.writer.kafka.KafkaWriter.

• sensorTopic : The kafka topic to that the parser will read messages from. If the topic is prefixed and suffixed by / then it is assumed to be a regex and will match any topic matching the pattern (e.g. /bro.*/ would match bro_cust0, bro_cust1 and bro_cust2)
• readMetadata : Boolean indicating whether to read metadata or not (The default is raw message strategy dependent). See below for a discussion about metadata.
• mergeMetadata : Boolean indicating whether to merge metadata with the message or not (The default is raw message strategy dependent). See below for a discussion about metadata.
• rawMessageStrategy : The strategy to use when reading the raw data and metadata. See below for a discussion about message reading strategies.
• rawMessageStrategyConfig : The raw message strategy configuration map. See below for a discussion about message reading strategies.
• parserConfig : A JSON Map representing the parser implementation specific configuration. Also include batch sizing and timeout for writer configuration here.
  • batchSize : Integer indicating number of records to batch together before sending to the writer. (default to 15)
  • batchTimeout : The timeout after which a batch will be flushed even if batchSize has not been met. Optional. If unspecified, or set to 0, it defaults to a system-determined duration which is a fraction of the Storm parameter topology.message.timeout.secs. Ignored if batchSize is 1, since this disables batching.
  • The kafka writer can be configured within the parser config as well. (This is all configured a priori, but this is convenient for overriding the settings). See here
• fieldTransformations : An array of complex objects representing the transformations to be done on the message generated from the parser before writing out to the kafka topic.
• securityProtocol : The security protocol to use for reading from kafka (this is a string). This can be overridden on the command line and also specified in the spout config via the security.protocol key. If both are specified, then they are merged and the CLI will take precedence. If multiple sensors are used, any non “PLAINTEXT” value will be used.
• cacheConfig : Cache config for stellar field transformations. This configures a least frequently used cache. This is a map with the following keys. If not explicitly configured (the default), then no cache will be used.
  • stellar.cache.maxSize - The maximum number of elements in the cache. Default is to not use a cache.
  • stellar.cache.maxTimeRetain - The maximum amount of time an element is kept in the cache (in minutes). Default is to not use a cache.

    Example of a cache config to contain at max 20000 stellar expressions for at most 20 minutes.:

    {
      "cacheConfig" : {
        "stellar.cache.maxSize" : 20000,
        "stellar.cache.maxTimeRetain" : 20
      }
    }
    

The fieldTransformations is a complex object which defines a transformation which can be done to a message. This transformation can

• Modify existing fields to a message
• Add new fields given the values of existing fields of a message
• Remove existing fields of a message

For platform specific configs, see the README of the appropriate project. This would include settings such as parallelism of individual components and general configuration.

• Storm

{
"customer_id" : "my_customer_id"
}

{
  "exception": "java.lang.IllegalStateException: Unable to parse Message: \"this is an invalid synthetic message\" }",
  "stack": "java.lang.IllegalStateException: Unable to parse Message: \"this is an invalid synthetic message\" ...\n",
  "raw_message": "\"this is an invalid synthetic message\" }",
  "error_hash": "3d498968e8df7f28d05db3037d4ad2a3a0095c22c14d881be45fac3f184dbcc3",
  "message": "Unable to parse Message: \"this is an invalid synthetic message\" }",
  "source.type": "error",
  "failed_sensor_type": "bro",
  "hostname": "node1",
  "error_type": "parser_error",
  "guid": "563d8d2a-1493-4758-be2f-5613bfd2d615",
  "timestamp": 1548366516634,
  "metron.metadata.topic": "bro",
  "metron.metadata.customer": "acme-inc"
}

 "fieldTransformations" : [
         { "transformation" : "STELLAR"
         ,"output" : [ "new_field", "old_field"]
         ,"config" : {
           "new_field" : "old_field"
          ,"old_field" : "null"
                     }
         }
 ]

 "fieldTransformations" : [
         { "transformation" : "STELLAR"
         ,"output" : [ "new_field"]
         ,"config" : {
           "new_field" : "TO_UPPER(field1)"
          ,"new_field" : "TO_LOWER(new_field)"
                     }
         }
 ]

{
...
    "fieldTransformations" : [
          {
           "transformation" : "STELLAR"
          ,"output" : [ "utc_timestamp", "url_host", "url_protocol" ]
          ,"config" : {
            "utc_timestamp" : "TO_EPOCH_TIMESTAMP(timestamp, 'yyyy-MM-dd
HH:mm:ss', MAP_GET(dc, dc2tz, 'UTC') )"
           ,"url_host" : "URL_TO_HOST(url)"
           ,"url_protocol" : "URL_TO_PROTOCOL(url)"
                      }
          }
                      ]
   ,"parserConfig" : {
      "dc2tz" : {
                "nyc" : "EST"
               ,"la" : "PST"
               ,"london" : "UTC"
                }
    }
}

{
  "parserClassName":"org.apache.metron.parsers.GrokParser",
  "sensorTopic":"yaf",
  "fieldTransformations" : [
                    {
                      "input" : "protocol"
                     ,"transformation": "IP_PROTOCOL"
                    }
                    ],
  "parserConfig":
  {
    "grokPath":"/patterns/yaf",
    "patternLabel":"YAF_DELIMITED",
    "timestampField":"start_time",
    "timeFields": ["start_time", "end_time"],
    "dateFormat":"yyyy-MM-dd HH:mm:ss.S"
  }
}

Parser Adapters

Parser adapters are loaded dynamically in each Metron topology. They are defined in the Parser Config (defined above) JSON file in Zookeeper.

Kafka Queue

The kafka queue associated with your parser is a collection point for all of the data sent to your parser. As such, make sure that the number of partitions in the kafka topic is sufficient to handle the throughput that you expect from your parser topology.

JSON Path

“JSONPath expressions always refer to a JSON structure in the same way as XPath expression are used in combination with an XML document.”

Stefan Goessner

• JSON Path concept
• Read about JSON Path library Apache Metron uses
• Try JSON Path expressions online