metron-streaming/Metron-Topologies/src/main/resources/Metron_Configs/topologies/bro/remote.yaml

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

name: "bro"
config:
    topology.workers: 1

components:
    -   id: "broParser"
        className: "org.apache.metron.parsing.parsers.BasicBroParser"
    -   id: "genericMessageFilter"
        className: "org.apache.metron.filters.GenericMessageFilter"
    -   id: "indexAdapter"
        className: "org.apache.metron.indexing.adapters.ESTimedRotatingAdapter"
    -   id: "metricConfig"
        className: "org.apache.commons.configuration.BaseConfiguration"
        configMethods:
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.reporter.graphite"
                    - "${org.apache.metron.metrics.reporter.graphite}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.reporter.console"
                    - "${org.apache.metron.metrics.reporter.console}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.reporter.jmx"
                    - "${org.apache.metron.metrics.reporter.jmx}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.graphite.address"
                    - "${org.apache.metron.metrics.graphite.address}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.graphite.port"
                    - "${org.apache.metron.metrics.graphite.port}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.TelemetryParserBolt.acks"
                    - "${org.apache.metron.metrics.TelemetryParserBolt.acks}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.TelemetryParserBolt.emits"
                    - "${org.apache.metron.metrics.TelemetryParserBolt.emits}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.TelemetryParserBolt.fails"
                    - "${org.apache.metron.metrics.TelemetryParserBolt.fails}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.GenericEnrichmentBolt.acks"
                    - "${org.apache.metron.metrics.GenericEnrichmentBolt.acks}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.GenericEnrichmentBolt.emits"
                    - "${org.apache.metron.metrics.GenericEnrichmentBolt.emits}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.GenericEnrichmentBolt.fails"
                    - "${org.apache.metron.metrics.GenericEnrichmentBolt.fails}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.TelemetryIndexingBolt.acks"
                    - "${org.apache.metron.metrics.TelemetryIndexingBolt.acks}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.TelemetryIndexingBolt.emits"
                    - "${org.apache.metron.metrics.TelemetryIndexingBolt.emits}"
            -   name: "setProperty"
                args:
                    - "org.apache.metron.metrics.TelemetryIndexingBolt.fails"
                    - "${org.apache.metron.metrics.TelemetryIndexingBolt.fails}"
    -   id: "zkHosts"
        className: "storm.kafka.ZkHosts"
        constructorArgs:
            - "${kafka.zk}"
    -   id: "kafkaConfig"
        className: "storm.kafka.SpoutConfig"
        constructorArgs:
            # zookeeper hosts
            - ref: "zkHosts"
            # topic name
            - "${spout.kafka.topic.bro}"
            # zk root
            - ""
            # id
            - "${spout.kafka.topic.bro}"
        properties:
            -   name: "forceFromStart"
                value: true
            -   name: "startOffsetTime"
                value: -1

spouts:
    -   id: "kafkaSpout"
        className: "storm.kafka.KafkaSpout"
        constructorArgs:
            - ref: "kafkaConfig"

bolts:
    -   id: "parserBolt"
        className: "org.apache.metron.bolt.TelemetryParserBolt"
        configMethods:
            -   name: "withMessageParser"
                args:
                    - ref: "broParser"
    -   id: "indexingBolt"
        className: "org.apache.metron.indexing.TelemetryIndexingBolt"
        configMethods:
            -   name: "withIndexIP"
                args:
                    - "${es.ip}"
            -   name: "withIndexPort"
                args:
                    - ${es.port}
            -   name: "withClusterName"
                args:
                    - "${es.clustername}"
            -   name: "withIndexName"
                args:
                    - "bro_index"
            -   name: "withIndexTimestamp"
                args:
                    - "yyyy.MM.dd.hh"
            -   name: "withDocumentName"
                args:
                    - "bro_doc"
            -   name: "withBulk"
                args:
                    - 1
            -   name: "withIndexAdapter"
                args:
                    - ref: "indexAdapter"
            -   name: "withMetricConfiguration"
                args:
                    - ref: "metricConfig"
    -   id: "errorIndexingBolt"
        className: "org.apache.metron.indexing.TelemetryIndexingBolt"
        configMethods:
            -   name: "withIndexIP"
                args:
                    - "${es.ip}"
            -   name: "withIndexPort"
                args:
                    - ${es.port}
            -   name: "withClusterName"
                args:
                    - "${es.clustername}"
            -   name: "withIndexName"
                args:
                    - "error"
            -   name: "withIndexTimestamp"
                args:
                    - "yyyy.MM"
            -   name: "withDocumentName"
                args:
                    - "bro_error"
            -   name: "withBulk"
                args:
                    - 1
            -   name: "withIndexAdapter"
                args:
                    - ref: "indexAdapter"
            -   name: "withMetricConfiguration"
                args:
                    - ref: "metricConfig"

streams:
    -   name: "spout -> parser"
        from: "kafkaSpout"
        to: "parserBolt"
        grouping:
            type: SHUFFLE
    -   name: "parser -> indexing"
        from: "parserBolt"
        to: "indexingBolt"
        grouping:
            streamId: "message"
            type: FIELDS
            args: ["key"]
    -   name: "parser -> errors"
        from: "parserBolt"
        to: "errorIndexingBolt"
        grouping:
            streamId: "error"
            type: SHUFFLE
    -   name: "indexing -> errors"
        from: "indexingBolt"
        to: "errorIndexingBolt"
        grouping:
            streamId: "error"
            type: SHUFFLE